Compare commits

...
42 Commits
Author SHA1 Message Date
KaifAhmad1 b382a7df6e chore: release version 0.2.4 2026-01-22 12:50:07 +05:30
Mohd Kaif b35081e015 Delete examples/demo_ontology_ingest.py 2026-01-21 18:27:06 +05:30
Mohd Kaif 7459393eea Merge pull request #214 from Hawksight-AI/ontology
feat(ontology): Implement OntologyIngestor and update exports
2026-01-21 13:51:54 +05:30
KaifAhmad1 b96e71ae72 feat(ontology): Implement OntologyIngestor and update exports
- Added OntologyIngestor in semantica/ingest/ontology_ingestor.py
- Updated semantica/ontology/__init__.py to export OntologyIngestor
- Updated semantica/ingest/methods.py to use OntologyIngestor
- Added tests for ontology ingestion
- Cleaned up temporary files
2026-01-21 13:46:46 +05:30
KaifAhmad1 fa8544c6d6 Release v0.2.3: Update version, changelog, and documentation 2026-01-20 12:08:46 +05:30
Mohd Kaif 87649b7422 Merge pull request #213 from Hawksight-AI/docs
Fix earnings call analysis notebook: attribute access and export logic
2026-01-20 01:52:42 +05:30
KaifAhmad1 d91619f191 Fix earnings call analysis notebook: attribute access and export logic 2026-01-20 01:51:29 +05:30
Mohd Kaif 064a0db7e6 Merge pull request #212 from Hawksight-AI/docs
Optimize Vector DB Storage in Earnings Call Analysis Notebook
2026-01-19 16:37:33 +05:30
KaifAhmad1 8214acc675 optimize vector db storage in earnings call analysis 2026-01-19 16:32:22 +05:30
Mohd Kaif 2bf55485ff Merge pull request #211 from Hawksight-AI/vector-store
Vector Store Performance Optimization
2026-01-19 13:40:44 +05:30
KaifAhmad1 1568237ce7 Add high-performance VectorStore ingestion and docs 2026-01-19 13:32:16 +05:30
Mohd Kaif f6c9d50e03 Merge pull request #210 from Hawksight-AI/docs
docs: update earnings call analysis notebook
2026-01-18 23:54:44 +05:30
KaifAhmad1 d9117b7c2f docs: update earnings call analysis notebook 2026-01-18 23:53:07 +05:30
Mohd Kaif 0eabfb861e Merge pull request #209 from Hawksight-AI/kg
Fix GraphBuilder External Relationships (#208, #206)
2026-01-18 22:13:16 +05:30
KaifAhmad1 9f77dfb761 Fix GraphBuilder external relationships; refs #208 #206 2026-01-18 22:10:02 +05:30
Mohd Kaif c990d09bd3 Merge pull request #205 from Hawksight-AI/docs
docs: changelog entry for JupyterLab progress flag (#181)
2026-01-17 17:14:54 +05:30
Mohd Kaif 9ebacf43c3 Update CHANGELOG.md 2026-01-17 17:09:41 +05:30
KaifAhmad1 7958ae78f6 docs: changelog entry for JupyterLab progress flag (#181) 2026-01-17 17:03:51 +05:30
Mohd Kaif 2c61fe6cda Merge pull request #204 from Hawksight-AI/utils
feat: allow disabling Jupyter progress output (#181)
2026-01-17 16:44:03 +05:30
KaifAhmad1 92b850ac26 feat: allow disabling Jupyter progress output (#181) 2026-01-17 16:40:15 +05:30
Mohd Kaif f7bd7016c5 Merge pull request #203 from Hawksight-AI/utils
Circular import between `pipeline_builder` and `pipeline_validator`
2026-01-17 16:12:02 +05:30
KaifAhmad1 8671385cbf fix: break pipeline circular import (#192, #193) and update changelog 2026-01-17 16:02:21 +05:30
Mohd Kaif b358acfabf Merge pull request #202 from Hawksight-AI/docs
Update Coockbook
2026-01-16 23:08:03 +05:30
KaifAhmad1 a39ec5fd20 Faster, class-based dedup: DuplicateDetector+EntityMerger with strict thresholds; build graph from deduplicated outputs; clean prints 2026-01-16 22:32:19 +05:30
KaifAhmad1 bbd6764215 Use deduplicated entities/relationships; optimize and clean deduplication; disable extra merging in GraphBuilder 2026-01-16 18:18:50 +05:30
KaifAhmad1 1b0b0551db Update Earnings Call Analysis notebook 2026-01-16 17:54:44 +05:30
Mohd Kaif a6b102fa3d Merge pull request #201 from don-simpson/feature/amazon-neptune-setup
feat: Added CloudFormation template and cookbook instructions for Amazon Neptune
2026-01-16 12:37:23 +05:30
Don Simpson 65d99f7f8a Added CloudFormation template that creates a dev cluster with a single [t3 instance](https://docs.aws.amazon.com/neptune/latest/userguide/manage-console-instances-t3.html) configured with a [public endpoint](https://docs.aws.amazon.com/neptune/latest/userguide/neptune-public-endpoints.html) and IAM Auth enabled (required for public endpoint), and creates an IAM User using least-privilege principles. See Get started with Neptune Database for free on the [Amazon Neptune pricing page](https://aws.amazon.com/neptune/pricing/).
Includes the CloudFormation template in the same directory as the [Amazon Neptune Cookbook](https://github.com/Hawksight-AI/semantica/blob/main/cookbook/introduction/21_Amazon_Neptune_Store.ipynb) and references it as a prerequisite in the cookbook.
2026-01-15 18:48:49 -05:00
Mohd Kaif 9b81137b26 Merge pull request #200 from Hawksight-AI/docs
Update earnings call analysis notebook with relation extraction fixes
2026-01-16 03:05:07 +05:30
KaifAhmad1 653523efeb Update earnings call analysis notebook with relation extraction fixes
- Update notebook to use corrected RelationExtractor API
- Move provider/model parameters to initialization
- Add verbose logging for debugging
- Include working relation extraction examples
2026-01-16 03:03:11 +05:30
Mohd Kaif ba04421d9b Merge pull request #199 from Hawksight-AI/docs
Update changelog for LLM relation extraction fixes
2026-01-16 02:59:44 +05:30
KaifAhmad1 5d3fe51dbd Update changelog for LLM relation extraction fixes
- Add comprehensive changelog entry for relation extraction parsing fixes
- Document breaking changes and new test coverage
- Update with provider normalization and JSON fallback details
2026-01-16 02:58:28 +05:30
Mohd Kaif f20782f517 Merge pull request #198 from Hawksight-AI/semantic-extract
Fix LLM Relation Extraction
2026-01-16 01:22:37 +05:30
KaifAhmad1 96dc5d754a Fix LLM relation extraction parsing and add tests
- Harden LLM relation extraction result handling to parse instructor/OpenAI/Groq variations
- Add structured JSON fallback when typed generation yields zero relations
- Strip acceptance of extra kwargs like max_tokens/max_entities_prompt in relation extraction internals
- Add comprehensive unit tests with mocked LLM provider
- Add integration tests for Groq provider with environment variable API key
- Ensure relation extraction completes and returns results when model identifies relations
2026-01-16 01:19:33 +05:30
Mohd Kaif cf84526cc7 Merge pull request #197 from Hawksight-AI/semantic-extract
Robust LLM Extraction and Groq 401 Fix
2026-01-15 22:46:20 +05:30
KaifAhmad1 5ad20abeab fix(semantic_extract): fix Groq 401 error and improve LLM provider robustness with instructor.from_provider 2026-01-15 22:43:11 +05:30
Mohd Kaif ade08a65ae Merge pull request #196 from Hawksight-AI/semantic-extract
Enhance RelationExtractor with core fixes and verbose logs
2026-01-15 19:03:36 +05:30
KaifAhmad1 fb25644fa7 Enhance RelationExtractor with core fixes and verbose logs
- Fix excessive entities being passed to LLM in RelationExtractor
- Add comprehensive 'Heartbeat' verbose logs to methods.py and providers.py
- Ensure robust API key handling and explicit error reporting
2026-01-15 19:00:42 +05:30
Mohd Kaif 63899f2427 Merge pull request #195 from Hawksight-AI/semantic-extract
Robust Semantic Extraction - API Key Handling & Error Reporting
2026-01-15 18:02:01 +05:30
KaifAhmad1 fd6e058275 feat(semantic_extract): enhance error reporting and API key robustness 2026-01-15 17:59:04 +05:30
Mohd Kaif 23d8207ef5 Merge pull request #194 from Hawksight-AI/semantic-extract
Robust API Key Handling in Semantic Extract Module
2026-01-15 16:32:52 +05:30
KaifAhmad1 f2a11fc8ad fix: robust api_key handling in semantic_extract module 2026-01-15 16:29:52 +05:30
42 changed files with 2937 additions and 261 deletions
+8 -1
View File
@@ -5,6 +5,7 @@ repos:
- id: trailing-whitespace
- id: end-of-file-fixer
- id: check-yaml
exclude: 'neptune-setup\.yaml$'
- id: check-json
- id: check-toml
- id: check-added-large-files
@@ -49,9 +50,15 @@ repos:
hooks:
- id: yamllint
args: ['-d', '{extends: default, rules: {line-length: {max: 120}}}']
exclude: 'neptune-setup\.yaml$'
- repo: https://github.com/aws-cloudformation/cfn-lint
rev: v1.43.3
hooks:
- id: cfn-lint
files: 'neptune-setup\.yaml$'
# Removed slow hooks for faster development:
# - mypy: Type checking (can be run manually or in CI)
# - bandit: Security scanning (can be run separately)
# - pytest: Testing (should be run manually, not on every commit)
+61
View File
@@ -7,6 +7,67 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
## [Unreleased]
## [0.2.4] - 2026-01-22
### Added
- **Ontology Ingestion Module**:
- Implemented `OntologyIngestor` in `semantica.ingest` for parsing RDF/OWL files (Turtle, RDF/XML, JSON-LD, N3) into standardized `OntologyData` objects.
- Added `ingest_ontology` convenience function and integrated it into the unified `ingest(source_type="ontology")` interface.
- Added recursive directory scanning support for batch ontology ingestion.
- Exposed ingestion tools in `semantica.ontology` for better discoverability.
- Added `OntologyData` dataclass for consistent metadata handling (source path, format, timestamps).
- **Documentation**:
- **Ontology Usage Guide**: Updated `ontology_usage.md` with comprehensive examples for single-file and directory ingestion.
- **API Reference**: Updated `ontology.md` with `OntologyIngestor` class documentation and method details.
- **Tests**:
- **Comprehensive Test Suite**: Added `tests/ingest/test_ontology_ingestor.py` covering all supported formats, error handling, and unified interface integration.
- **Demo Script**: Added `examples/demo_ontology_ingest.py` for end-to-end usage demonstration.
## [0.2.3] - 2026-01-20
### Fixed
- **LLM Relation Extraction Parsing**:
- Fixed relation extraction returning zero relations despite successful API calls to Groq and other providers
- Normalized typed responses from instructor/OpenAI/Groq to consistent dict format before parsing
- Added structured JSON fallback when typed generation yields zero relations to avoid silent empty outputs
- Removed acceptance of extra kwargs (`max_tokens`, `max_entities_prompt`) from relation extraction internals
- Filtered kwargs passed to provider LLM calls to only `temperature` and `verbose`
- **API Parameter Handling**:
- Limited kwargs forwarded in chunked extraction helper to prevent parameter leakage
- Ensured minimal, safe parameters are passed to provider calls
- **Pipeline Circular Import (Issues #192, #193)**:
- Fixed circular import between `pipeline_builder` and `pipeline_validator` triggered during `semantica.pipeline` import
- Lazy-loaded `PipelineValidator` inside `PipelineBuilder.__init__` and guarded type hints with `TYPE_CHECKING`
- Ensured `from semantica.deduplication import DuplicateDetector` no longer fails even when pipeline module is imported
- **JupyterLab Progress Output (Issue #181)**:
- Added `SEMANTICA_DISABLE_JUPYTER_PROGRESS` environment variable to disable rich Jupyter/Colab progress tables
- When enabled, progress falls back to console-style output, preventing infinite scrolling and JupyterLab out-of-memory errors
### Added
- **Comprehensive Test Suite**:
- - Added unit tests (`tests/test_relations_llm.py`) with mocked LLM provider covering both typed and structured response paths
- - Added integration tests (`tests/integration/test_relations_groq.py`) for real Groq API calls with environment variable API key
- - Tests validate relation extraction completion and result parsing across different response formats
- **Amazon Neptune Dev Environment**:
- - Added CloudFormation template (`cookbook/introduction/neptune-setup.yaml`) to provision a dev Neptune cluster with public endpoint and IAM auth enabled
- - Documented deployment, cost estimates, and IAM User vs IAM Role best practices in `cookbook/introduction/21_Amazon_Neptune_Store.ipynb`
- - Added `cfn-lint` to `.pre-commit-config.yaml` for validating CloudFormation templates while excluding `neptune-setup.yaml` from generic YAML linters
- **Vector Store High-Performance Ingestion**:
- - Added `VectorStore.add_documents` for high-throughput ingestion with automatic embedding generation, batching, and parallel processing
- - Added `VectorStore.embed_batch` helper for generating embeddings for lists of texts without immediately storing them
- - Enabled default parallel ingestion in `VectorStore` with `max_workers=6` for common workloads
- - Added dedicated documentation page `docs/vector_store_usage.md` describing high-performance vector store usage and configuration
- - Added `tests/vector_store/test_vector_store_parallel.py` covering parallel vs sequential performance, error handling, and edge cases for `add_documents` and `embed_batch`
### Changed
- **Relation Extraction API**:
- - Simplified parameter interface by removing unused kwargs that were previously ignored
- - Improved error handling and verbose logging for debugging relation extraction issues
- - Enhanced robustness of post-response parsing across different LLM providers
- **Vector Store Defaults and Examples**:
- - Standardized `VectorStore` default concurrency to `max_workers=6` for parallel ingestion
- - Updated vector store reference documentation and usage guides to rely on implicit defaults instead of requiring manual `max_workers` configuration in examples
## [0.2.2] - 2026-01-15
+1 -1
View File
@@ -28,7 +28,7 @@
*The missing fabric between raw data and AI engineering. A comprehensive open-source framework for building semantic layers and knowledge engineering systems that transform unstructured data into AI-ready knowledge — powering Knowledge Graph-Powered RAG (GraphRAG), AI Agents, Multi-Agent Systems, and AI applications with structured semantic knowledge.*
**100% Open Source****MIT Licensed****Latest Version: 0.2.2****Production Ready****Community Driven**
**100% Open Source****MIT Licensed****Latest Version: 0.2.4****Production Ready****Community Driven**
[**Discord**](https://discord.gg/pMHguUzG)
+3 -3
View File
@@ -26,10 +26,10 @@ Before releasing, ensure:
The project uses GitHub Actions for automated releases to PyPI.
1. **Tag the commit**: Create a new git tag for the version (e.g., `v0.2.2`).
1.29. **Tag the commit**: Create a new git tag for the version (e.g., `v0.2.3`).
```bash
git tag -a v0.2.2 -m "Release v0.2.2"
git push origin v0.2.2
git tag -a v0.2.3 -m "Release v0.2.3"
git push origin v0.2.3
```
2. **GitHub Action**: The `Release` workflow will automatically trigger, build the package, create a GitHub Release, and publish to PyPI using Trusted Publishing.
+1
View File
@@ -6,6 +6,7 @@ We actively support the following versions of Semantica with security updates:
| Version | Supported |
| ------- | ------------------ |
| 0.2.3 | :white_check_mark: |
| 0.2.2 | :white_check_mark: |
| 0.2.1 | :white_check_mark: |
| 0.2.0 | :white_check_mark: |
@@ -25,6 +25,57 @@
"- AWS credentials configured (boto3, environment variables, or IAM role)\n",
"- Network access to your Neptune cluster (VPC, security groups)\n",
"\n",
"#### Quick Setup with CloudFormation\n",
"\n",
"If you don't have a Neptune cluster, use the provided CloudFormation template to create one with a public endpoint and IAM authentication:\n",
"\n",
"```bash\n",
"# Deploy the Neptune stack (takes ~15-20 minutes)\n",
"aws cloudformation create-stack \\\n",
" --stack-name semantica-neptune \\\n",
" --template-body file://neptune-setup.yaml \\\n",
" --capabilities CAPABILITY_NAMED_IAM\n",
"\n",
"# Wait for stack creation to complete\n",
"aws cloudformation wait stack-create-complete --stack-name semantica-neptune\n",
"\n",
"# Get the outputs (endpoint, port, credentials)\n",
"aws cloudformation describe-stacks --stack-name semantica-neptune \\\n",
" --query 'Stacks[0].Outputs' --output table\n",
"```\n",
"\n",
"The template creates:\n",
"- VPC with public subnets and Internet Gateway\n",
"- Neptune cluster (`db.t3.medium`) with IAM authentication enabled\n",
"- IAM user with least-privilege access for OpenCypher queries\n",
"- Security group allowing Bolt protocol (port 8182) access\n",
"\n",
"> ⚠️ **Security Note**: This template creates an IAM User with static access keys for simplicity in demo/test environments. For production use, we recommend IAM Roles (EC2 instance roles, ECS task roles, Lambda execution roles) which provide temporary credentials that are automatically rotated. The secret access key in the Cloudformation outputs is provided in plaintext to simplify initial setup - in production, use AWS Secrets Manager.\n",
"\n",
"**Outputs:**\n",
"- `NeptuneEndpoint` - Cluster hostname (use as `NEPTUNE_ENDPOINT`)\n",
"- `NeptunePort` - 8182 (use as `NEPTUNE_PORT`)\n",
"- `AwsAccessKeyId` - IAM user access key (use as `AWS_ACCESS_KEY_ID`)\n",
"- `AwsSecretAccessKey` - IAM user secret key in **plaintext** (use as `AWS_SECRET_ACCESS_KEY`)\n",
"- `AwsRegion` - Deployment region (use as `AWS_REGION`)\n",
"\n",
"**Cleanup:**\n",
"```bash\n",
"aws cloudformation delete-stack --stack-name semantica-neptune\n",
"```\n",
"\n",
"**Estimated Monthly Cost (approximately 100-105 USD/month at 100% utilization):**\n",
"\n",
"| Resource | Cost (USD) |\n",
"| --- | --- |\n",
"| Neptune db.t3.medium instance | ~96/month (0.132/hr) |\n",
"| Storage (10 GB) | ~1/month |\n",
"| I/O requests | ~1-5/month |\n",
"| Public IPv4 address | ~3.60/month (0.005/hr) |\n",
"| VPC, subnets, route tables, Internet Gateway, IAM | No Additional Charge |\n",
"\n",
"> **Free Tier**: New Neptune users get 30 days free (750 hours of db.t3.medium, 10M I/Os, 1 GB storage). Delete the stack when not in use to avoid charges.\n",
"\n",
"---"
]
},
@@ -70,14 +121,21 @@
"import os\n",
"\n",
"# Neptune cluster configuration - REPLACE WITH YOUR VALUES\n",
"# (Get these from CloudFormation stack outputs)\n",
"os.environ[\"NEPTUNE_ENDPOINT\"] = \"your-cluster.us-east-1.neptune.amazonaws.com\"\n",
"os.environ[\"NEPTUNE_PORT\"] = \"8182\"\n",
"os.environ[\"AWS_REGION\"] = \"us-east-1\"\n",
"\n",
"# AWS credentials (if using IAM Auth and not relying on IAM role or ~/.aws/credentials)\n",
"# os.environ[\"AWS_ACCESS_KEY_ID\"] = \"your-access-key-id\"\n",
"# os.environ[\"AWS_SECRET_ACCESS_KEY\"] = \"your-secret-access-key\"\n",
"# os.environ[\"AWS_SESSION_TOKEN\"] = \"your-session-token\"\n",
"# AWS credentials for IAM Authentication\n",
"# Option 1: IAM User (static credentials from CloudFormation template)\n",
"# os.environ[\"AWS_ACCESS_KEY_ID\"] = \"AKIA...\" # From AwsAccessKeyId output\n",
"# os.environ[\"AWS_SECRET_ACCESS_KEY\"] = \"...\" # From AwsSecretAccessKey output\n",
"# Note: No AWS_SESSION_TOKEN needed for IAM users\n",
"\n",
"# Option 2: IAM Role / Temporary credentials (e.g., STS AssumeRole, EC2 instance role)\n",
"# os.environ[\"AWS_ACCESS_KEY_ID\"] = \"ASIA...\" # Temporary access key\n",
"# os.environ[\"AWS_SECRET_ACCESS_KEY\"] = \"...\" # Temporary secret key\n",
"# os.environ[\"AWS_SESSION_TOKEN\"] = \"...\" # REQUIRED for temporary credentials\n",
"\n",
"print(f\"Neptune Endpoint: {os.environ.get('NEPTUNE_ENDPOINT')}\")\n",
"print(f\"AWS Region: {os.environ.get('AWS_REGION')}\")"
+228
View File
@@ -0,0 +1,228 @@
AWSTemplateFormatVersion: '2010-09-09'
Description: >
Amazon Neptune cluster with public endpoint, IAM authentication, and least-privilege
IAM user for Semantica cookbook. Uses db.t3.medium (most cost-effective Neptune instance type).
Parameters:
EnvironmentName:
Type: String
Default: semantica-neptune
Description: Environment name prefix for resource naming
Resources:
# =============================================================================
# VPC & NETWORKING
# =============================================================================
VPC:
Type: AWS::EC2::VPC
Properties:
CidrBlock: 10.0.0.0/16
EnableDnsHostnames: true
EnableDnsSupport: true
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-vpc
InternetGateway:
Type: AWS::EC2::InternetGateway
Properties:
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-igw
InternetGatewayAttachment:
Type: AWS::EC2::VPCGatewayAttachment
Properties:
InternetGatewayId: !Ref InternetGateway
VpcId: !Ref VPC
PublicSubnet1:
Type: AWS::EC2::Subnet
Properties:
VpcId: !Ref VPC
AvailabilityZone: !Select [0, !GetAZs '']
CidrBlock: 10.0.1.0/24
MapPublicIpOnLaunch: true
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-public-subnet-1
PublicSubnet2:
Type: AWS::EC2::Subnet
Properties:
VpcId: !Ref VPC
AvailabilityZone: !Select [1, !GetAZs '']
CidrBlock: 10.0.2.0/24
MapPublicIpOnLaunch: true
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-public-subnet-2
PublicRouteTable:
Type: AWS::EC2::RouteTable
Properties:
VpcId: !Ref VPC
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-public-rt
DefaultPublicRoute:
Type: AWS::EC2::Route
DependsOn: InternetGatewayAttachment
Properties:
RouteTableId: !Ref PublicRouteTable
DestinationCidrBlock: 0.0.0.0/0
GatewayId: !Ref InternetGateway
PublicSubnet1RouteTableAssociation:
Type: AWS::EC2::SubnetRouteTableAssociation
Properties:
RouteTableId: !Ref PublicRouteTable
SubnetId: !Ref PublicSubnet1
PublicSubnet2RouteTableAssociation:
Type: AWS::EC2::SubnetRouteTableAssociation
Properties:
RouteTableId: !Ref PublicRouteTable
SubnetId: !Ref PublicSubnet2
# =============================================================================
# SECURITY GROUP
# =============================================================================
NeptuneSecurityGroup:
Type: AWS::EC2::SecurityGroup
Properties:
GroupName: !Sub ${EnvironmentName}-neptune-sg
GroupDescription: Security group for Neptune cluster - allows Bolt protocol access
VpcId: !Ref VPC
SecurityGroupIngress:
- IpProtocol: tcp
FromPort: 8182
ToPort: 8182
CidrIp: 0.0.0.0/0
Description: Allow Bolt protocol access from anywhere
SecurityGroupEgress:
- IpProtocol: -1
CidrIp: 0.0.0.0/0
Description: Allow all outbound traffic
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-neptune-sg
# =============================================================================
# NEPTUNE CLUSTER
# =============================================================================
NeptuneSubnetGroup:
Type: AWS::Neptune::DBSubnetGroup
Properties:
DBSubnetGroupDescription: Subnet group for Neptune cluster
DBSubnetGroupName: !Sub ${EnvironmentName}-subnet-group
SubnetIds:
- !Ref PublicSubnet1
- !Ref PublicSubnet2
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-subnet-group
NeptuneCluster:
Type: AWS::Neptune::DBCluster
Properties:
DBClusterIdentifier: !Sub ${EnvironmentName}-cluster
DBSubnetGroupName: !Ref NeptuneSubnetGroup
VpcSecurityGroupIds:
- !Ref NeptuneSecurityGroup
EngineVersion: '1.4.6.3'
IamAuthEnabled: true
StorageEncrypted: true
DeletionProtection: false
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-cluster
NeptuneInstance:
Type: AWS::Neptune::DBInstance
Properties:
DBInstanceIdentifier: !Sub ${EnvironmentName}-instance
DBInstanceClass: db.t3.medium
DBClusterIdentifier: !Ref NeptuneCluster
PubliclyAccessible: true
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-instance
# =============================================================================
# IAM USER WITH LEAST PRIVILEGES
# =============================================================================
NeptuneUser:
Type: AWS::IAM::User
Properties:
UserName: !Sub ${EnvironmentName}-user
Tags:
- Key: Name
Value: !Sub ${EnvironmentName}-user
NeptuneUserPolicy:
Type: AWS::IAM::Policy
Properties:
PolicyName: !Sub ${EnvironmentName}-neptune-access
Users:
- !Ref NeptuneUser
PolicyDocument:
Version: '2012-10-17'
Statement:
- Sid: NeptuneDataAccess
Effect: Allow
Action:
- neptune-db:connect
- neptune-db:ReadDataViaQuery
- neptune-db:WriteDataViaQuery
- neptune-db:DeleteDataViaQuery
Resource: !Sub
- arn:aws:neptune-db:${AWS::Region}:${AWS::AccountId}:${ClusterResourceId}/*
- ClusterResourceId: !GetAtt NeptuneCluster.ClusterResourceId
NeptuneUserAccessKey:
Type: AWS::IAM::AccessKey
Properties:
UserName: !Ref NeptuneUser
# =============================================================================
# OUTPUTS
# =============================================================================
Outputs:
NeptuneEndpoint:
Description: Neptune cluster endpoint (hostname only) - use as NEPTUNE_ENDPOINT
Value: !GetAtt NeptuneCluster.Endpoint
NeptunePort:
Description: Neptune cluster port - use as NEPTUNE_PORT
Value: !GetAtt NeptuneCluster.Port
AwsAccessKeyId:
Description: Access key ID for the Neptune IAM user - use as AWS_ACCESS_KEY_ID
Value: !Ref NeptuneUserAccessKey
AwsSecretAccessKey:
Description: Secret access key for the Neptune IAM user - use as AWS_SECRET_ACCESS_KEY
Value: !GetAtt NeptuneUserAccessKey.SecretAccessKey
AwsRegion:
Description: AWS region where Neptune is deployed - use as AWS_REGION
Value: !Ref AWS::Region
NeptuneClusterResourceId:
Description: Neptune cluster resource ID (for IAM policy reference)
Value: !GetAtt NeptuneCluster.ClusterResourceId
VpcId:
Description: VPC ID
Value: !Ref VPC
SecurityGroupId:
Description: Neptune security group ID
Value: !Ref NeptuneSecurityGroup
@@ -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",
@@ -385,22 +389,79 @@
" \"FOR_PERIOD\",\n",
" \"RELATED_TO\",\n",
" ],\n",
" provider=\"groq\",\n",
" llm_model=\"llama-3.1-8b-instant\",\n",
" api_key=GROQ_API_KEY,\n",
" temperature=0.0,\n",
" verbose=False,\n",
")\n",
"\n",
"relationships = [\n",
" r\n",
" for c in chunks\n",
" for r in relation_extractor.extract_relations(\n",
" text=get_chunk_text(c),\n",
" entities=all_entities,\n",
" provider=\"groq\",\n",
" llm_model=\"llama-3.1-8b-instant\",\n",
" temperature=0.0,\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",
" if get_chunk_text(c).strip()\n",
"]\n",
"print(\"Relationships:\", len(relationships))\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",
"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",
" try:\n",
" rels = future.result(timeout=CHUNK_TIMEOUT)\n",
" relationships.extend(rels)\n",
" print(f\" relations={len(rels)}\")\n",
"\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",
"if relationships:\n",
" for r in relationships[:10]:\n",
" print(f\"{r.subject.text} → {r.predicate} → {r.object.text}\")\n"
]
},
{
@@ -421,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])"
]
},
{
@@ -478,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])"
]
},
{
@@ -510,38 +617,80 @@
"outputs": [],
"source": [
"from semantica.deduplication import DuplicateDetector, EntityMerger\n",
"import time\n",
"\n",
"duplicate_detector = DuplicateDetector(\n",
" similarity_threshold=0.8,\n",
" confidence_threshold=0.7,\n",
"start_time = time.time()\n",
"\n",
"raw = []\n",
"for i, e in enumerate(entities):\n",
" raw.append({\n",
" \"id\": getattr(e, \"id\", None) or f\"entity_{i}_{getattr(e, 'text', str(e))}\",\n",
" \"name\": (getattr(e, \"text\", getattr(e, \"name\", \"\")) or \"\").strip(),\n",
" \"type\": getattr(e, \"label\", \"UNKNOWN\"),\n",
" \"confidence\": float(getattr(e, \"confidence\", 1.0) or 1.0),\n",
" \"metadata\": getattr(e, \"metadata\", {}),\n",
" })\n",
"\n",
"filtered = [r for r in raw if r[\"name\"] and len(r[\"name\"]) >= 3]\n",
"\n",
"collapsed = {}\n",
"for ent in filtered:\n",
" key = (ent[\"type\"], ent[\"name\"].lower())\n",
" best = collapsed.get(key)\n",
" if best is None or ent[\"confidence\"] > best[\"confidence\"]:\n",
" collapsed[key] = ent\n",
"\n",
"entity_dicts = list(collapsed.values())\n",
"\n",
"detector = DuplicateDetector(\n",
" similarity_threshold=0.96,\n",
" confidence_threshold=0.92,\n",
" use_clustering=True,\n",
")\n",
"\n",
"entity_dicts = [\n",
" {\n",
" \"id\": getattr(e, \"id\", \"\"),\n",
" \"name\": getattr(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",
"]\n",
"detector.detect_duplicate_groups(entity_dicts)\n",
"\n",
"duplicates = duplicate_detector.detect_duplicates(entity_dicts)\n",
"merger = EntityMerger(\n",
" preserve_provenance=True,\n",
" detector={\n",
" \"similarity_threshold\": 0.96,\n",
" \"confidence_threshold\": 0.92,\n",
" \"use_clustering\": True,\n",
" },\n",
" strategy={\"default_strategy\": \"keep_most_complete\"},\n",
")\n",
"\n",
"entity_merger = EntityMerger(preserve_provenance=True)\n",
"\n",
"merge_operations = entity_merger.merge_duplicates(\n",
" entity_dicts,\n",
"merge_operations = merger.merge_duplicates(\n",
" entities=entity_dicts,\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] or entity_dicts\n",
"\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"
"entity_id_mapping = {}\n",
"for op in merge_operations:\n",
" mid = op.merged_entity[\"id\"]\n",
" for sid in op.source_ids:\n",
" entity_id_mapping[sid] = mid\n",
"\n",
"deduplicated_relationships = []\n",
"for rel in normalized_relationships:\n",
" s = entity_id_mapping.get(rel[\"source_id\"], rel[\"source_id\"])\n",
" t = entity_id_mapping.get(rel[\"target_id\"], rel[\"target_id\"])\n",
" if s != t:\n",
" r = rel.copy()\n",
" r[\"source_id\"], r[\"target_id\"] = s, t\n",
" deduplicated_relationships.append(r)\n",
"\n",
"print({\n",
" \"time_seconds\": round(time.time() - start_time, 2),\n",
" \"entities_in\": len(raw),\n",
" \"entities_after_filter\": len(filtered),\n",
" \"entities_after_exact\": len(entity_dicts),\n",
" \"entities_out\": len(deduplicated_entities),\n",
" \"duplicates_removed\": len(entity_dicts) - len(deduplicated_entities),\n",
" \"relationships_updated\": len(deduplicated_relationships),\n",
"})"
]
},
{
@@ -561,28 +710,17 @@
"source": [
"from semantica.kg import GraphBuilder\n",
"\n",
"# Deduplication is already done; avoid additional entity resolution/merging\n",
"graph_builder = GraphBuilder(\n",
" merge_entities=True,\n",
" entity_resolution_strategy=\"fuzzy\",\n",
" merge_entities=False,\n",
" entity_resolution_strategy=\"none\",\n",
")\n",
"\n",
"triplet_relationships = [\n",
" {\n",
" \"source\": t.subject,\n",
" \"predicate\": t.predicate,\n",
" \"target\": t.object,\n",
" \"confidence\": t.confidence,\n",
" \"metadata\": t.metadata,\n",
" }\n",
" for t in validated_triplets\n",
"]\n",
"\n",
"final_relationships = resolved_relationships + triplet_relationships\n",
"final_relationships = deduplicated_relationships\n",
"\n",
"kg_data = {\n",
" \"entities\": merged_entities,\n",
" \"entities\": deduplicated_entities,\n",
" \"relationships\": final_relationships,\n",
" \"triplets\": validated_triplets,\n",
" \"metadata\": {\n",
" \"source\": \"earnings_call_transcript\",\n",
" \"extraction_method\": \"Groq LLM\",\n",
@@ -591,12 +729,12 @@
"\n",
"knowledge_graph = graph_builder.build(\n",
" sources=[kg_data],\n",
" merge_entities=True,\n",
" merge_entities=False,\n",
")\n",
"\n",
"print(\"Knowledge graph build completed\")\n",
"print(\"Knowledge graph build completed (no additional merging)\")\n",
"print(\"Final entities:\", len(knowledge_graph.get(\"entities\", [])))\n",
"print(\"Final relationships:\", len(knowledge_graph.get(\"relationships\", [])))\n"
"print(\"Final relationships:\", len(knowledge_graph.get(\"relationships\", [])))"
]
},
{
@@ -641,12 +779,10 @@
"connectivity = graph_analyzer.analyze_connectivity(knowledge_graph)\n",
"metrics = graph_analyzer.compute_metrics(knowledge_graph)\n",
"\n",
"top_entities = centrality.get(\"rankings\", [])[:5]\n",
"num_communities = len(communities.get(\"communities\", []))\n",
"\n",
"print(\"Graph analysis completed\")\n",
"print(\"Communities:\", num_communities)\n",
"print(\"Top entities:\", len(top_entities))\n"
"print(\"Communities:\", num_communities)"
]
},
{
@@ -779,31 +915,62 @@
"metadata": {},
"outputs": [],
"source": [
"import time\n",
"from semantica.vector_store import VectorStore\n",
"from semantica.context import ContextRetriever\n",
"\n",
"vector_store = VectorStore(backend=\"faiss\")\n",
"if 'chunks' not in locals() or not chunks:\n",
" raise ValueError(\"Chunks not found. Please run Step 3 first.\")\n",
"\n",
"vector_store.add(\n",
" texts=[parsed_doc[\"full_text\"]],\n",
" metadata=[{\"source\": \"earnings_call\", \"type\": \"transcript\"}],\n",
"# Extract text content safely\n",
"chunk_texts = [getattr(c, \"content\", getattr(c, \"text\", \"\")) for c in chunks]\n",
"chunk_metadatas = [\n",
" {\n",
" \"source\": \"earnings_call\", \n",
" \"type\": \"transcript\", \n",
" \"chunk_index\": i,\n",
" **(getattr(c, \"metadata\", {}) or {})\n",
" }\n",
" for i, c in enumerate(chunks)\n",
"]\n",
"\n",
"# Initialize Vector Store (Optimized for Speed)\n",
"# dimension=384 matches the default fast model (BAAI/bge-small-en-v1.5)\n",
"vector_store = VectorStore(\n",
" backend=\"faiss\", \n",
" dimension=384, \n",
" max_workers=16\n",
")\n",
"\n",
"print(f\"Storing {len(chunks)} chunks with high-performance settings...\")\n",
"start_time = time.time()\n",
"\n",
"# Store in large batches with parallel processing\n",
"vector_ids = vector_store.add_documents(\n",
" documents=chunk_texts,\n",
" metadata=chunk_metadatas,\n",
" batch_size=128,\n",
" parallel=True\n",
")\n",
"\n",
"print(f\"✅ Stored in {time.time() - start_time:.2f}s\")\n",
"\n",
"# Initialize Hybrid Retriever\n",
"context_retriever = ContextRetriever(\n",
" knowledge_graph=knowledge_graph,\n",
" knowledge_graph=knowledge_graph, # Assumes knowledge_graph exists\n",
" vector_store=vector_store,\n",
" hybrid_alpha=0.6,\n",
" use_graph_expansion=True,\n",
" max_expansion_hops=2,\n",
")\n",
"\n",
"# Test Retrieval\n",
"queries = [\n",
" \"What was the company's revenue guidance?\",\n",
" \"What were the key financial metrics discussed?\",\n",
"]\n",
"\n",
"retrieved_contexts = []\n",
"\n",
"for query in queries:\n",
" results = context_retriever.retrieve(\n",
" query=query,\n",
@@ -813,8 +980,7 @@
" retrieved_contexts.append(results)\n",
"\n",
"print(\"Hybrid GraphRAG configured\")\n",
"print(\"Queries processed:\", len(queries))\n",
"print(\"Sample results:\", len(retrieved_contexts[0]) if retrieved_contexts else 0)\n"
"print(\"Sample results:\", len(retrieved_contexts[0]) if retrieved_contexts else 0)"
]
},
{
@@ -857,10 +1023,13 @@
" retention_days=30,\n",
")\n",
"\n",
"entity_count = len(knowledge_graph.get(\"entities\", []))\n",
"relationship_count = len(knowledge_graph.get(\"relationships\", []))\n",
"\n",
"memory_contents = [\n",
" f\"Earnings call transcript: {parsed_doc['metadata'].get('title', 'Earnings Call')}\",\n",
" f\"Financial metrics extracted: {sum(len(v) for v in financial_metrics.values())}\",\n",
" f\"Key entities identified: {len(merged_entities)}\",\n",
" f\"Graph entities: {entity_count}\",\n",
" f\"Graph relationships: {relationship_count}\",\n",
"]\n",
"\n",
"memory_ids = []\n",
@@ -885,7 +1054,7 @@
"print(\"Agent memory configured\")\n",
"print(\"Memories stored:\", len(memory_ids))\n",
"print(\"Total memories:\", memory_stats.get(\"total_memories\", 0))\n",
"print(\"Retrieved memories:\", len(financial_memories))\n"
"print(\"Retrieved memories:\", len(financial_memories))"
]
},
{
@@ -936,7 +1105,7 @@
},
{
"cell_type": "code",
"execution_count": 4,
"execution_count": null,
"metadata": {},
"outputs": [
{
@@ -964,7 +1133,7 @@
")\n",
"\n",
"memory_id = agent_context.store(\n",
" content=parsed_doc[\"full_text\"][:1000],\n",
" content=chunks,\n",
" metadata={\"source\": \"earnings_call\", \"date\": \"2024-Q1\"},\n",
" extract_entities=True,\n",
" extract_relationships=True,\n",
@@ -1008,32 +1177,71 @@
"\n",
"generated_answers = []\n",
"\n",
"print(\"--- Generating Enhanced Answers ---\\n\")\n",
"\n",
"def format_context(retrieved_contexts):\n",
" \"\"\"Formats retrieved context with graph information.\"\"\"\n",
" formatted_parts = []\n",
" \n",
" for i, ctx in enumerate(retrieved_contexts):\n",
" content = getattr(ctx, \"content\", \"\")\n",
" source = getattr(ctx, \"source\", \"unknown\")\n",
" \n",
" # Format related entities from the graph\n",
" related_entities = getattr(ctx, \"related_entities\", [])\n",
" entities_str = \", \".join([\n",
" f\"{e.get('name', 'Unknown')} ({e.get('type', 'Entity')})\" \n",
" for e in related_entities[:5] # Limit to top 5 per chunk\n",
" ])\n",
" \n",
" # Format related relationships\n",
" related_rels = getattr(ctx, \"related_relationships\", [])\n",
" rels_str = \"; \".join([\n",
" f\"{r.get('source', '')} -> {r.get('type', '')} -> {r.get('target', '')}\"\n",
" for r in related_rels[:3] # Limit to top 3 per chunk\n",
" ])\n",
" \n",
" part = f\"Source {i+1} ({source}):\\n{content}\\n\"\n",
" if entities_str:\n",
" part += f\"Related Entities: {entities_str}\\n\"\n",
" if rels_str:\n",
" part += f\"Graph Connections: {rels_str}\\n\"\n",
" \n",
" formatted_parts.append(part)\n",
" \n",
" return \"\\n---\\n\".join(formatted_parts)\n",
"\n",
"for question in financial_questions:\n",
" print(f\"Question: {question}\")\n",
" \n",
" # Retrieve with graph expansion enabled and higher limits\n",
" retrieved_contexts = context_retriever.retrieve(\n",
" query=question,\n",
" max_results=3,\n",
" max_results=10, # Increased from 3\n",
" min_relevance_score=0.2,\n",
" use_graph_expansion=True, # Explicitly enable graph expansion\n",
" max_hops=2 # Traverse up to 2 hops in the graph\n",
" )\n",
"\n",
" context_text = \"\\n\\n\".join(\n",
" ctx.get(\"content\", ctx.get(\"text\", \"\"))\n",
" for ctx in retrieved_contexts\n",
" )[:1000]\n",
" # Use the rich formatter\n",
" context_text = format_context(retrieved_contexts)\n",
"\n",
" entity_names = [\n",
" entity.get(\"name\", \"\")\n",
" for entity in knowledge_graph.get(\"entities\", [])[:5]\n",
" # Get global key entities (optional, but good for high-level context)\n",
" global_entities = [\n",
" f\"{e.get('name', '')} ({e.get('type', '')})\"\n",
" for e in knowledge_graph.get(\"entities\", [])[:10]\n",
" ]\n",
" entities_text = \", \".join(entity_names) or \"N/A\"\n",
" global_entities_text = \", \".join(global_entities)\n",
"\n",
" prompt = f\"\"\"\n",
"Answer the question using only the context below.\n",
"Answer the question comprehensively using the provided context.\n",
"The context includes text chunks and knowledge graph connections (entities and relationships).\n",
"If the answer is not present, say so.\n",
"\n",
"Context:\n",
"{context_text}\n",
"\n",
"Key entities: {entities_text}\n",
"Global Key Entities: {global_entities_text}\n",
"\n",
"Question:\n",
"{question}\n",
@@ -1044,16 +1252,18 @@
" try:\n",
" answer = groq_llm.generate(\n",
" prompt,\n",
" temperature=0.7,\n",
" max_tokens=400,\n",
" temperature=0.3, # Lower temperature for more factual answers\n",
" max_tokens=1000, # Allow longer answers\n",
" )\n",
" except Exception as error:\n",
" answer = f\"Answer generation failed: {error}\"\n",
"\n",
" generated_answers.append(answer)\n",
" print(f\"Answer: {answer}\\n\")\n",
" print(\"-\" * 50 + \"\\n\")\n",
"\n",
"print(\"Answer generation completed\")\n",
"print(\"Questions answered:\", len(generated_answers))\n"
"print(\"Questions answered:\", len(generated_answers))"
]
},
{
@@ -1072,28 +1282,44 @@
"outputs": [],
"source": [
"from semantica.export import JSONExporter, RDFExporter\n",
"import json\n",
"\n",
"# Initialize exporters\n",
"json_exporter = JSONExporter()\n",
"rdf_exporter = RDFExporter()\n",
"\n",
"kg_json = json_exporter.export(knowledge_graph, format=\"json\")\n",
"kg_rdf = rdf_exporter.export_to_rdf(knowledge_graph, format=\"turtle\")\n",
"# Define output file paths\n",
"json_output_path = \"knowledge_graph.json\"\n",
"rdf_output_path = \"knowledge_graph.ttl\"\n",
"\n",
"# Export to files (required by the API)\n",
"json_exporter.export(knowledge_graph, file_path=json_output_path, format=\"json\")\n",
"\n",
"# FIXED: Use .export() instead of .export_to_rdf() to write to disk\n",
"rdf_exporter.export(knowledge_graph, file_path=rdf_output_path, format=\"turtle\")\n",
"\n",
"# Load the RDF file content to check its size\n",
"with open(rdf_output_path, \"r\", encoding=\"utf-8\") as f:\n",
" kg_rdf_content = f.read()\n",
"\n",
"# Create analysis summary\n",
"analysis_summary = {\n",
" \"entities\": len(knowledge_graph.get(\"entities\", [])),\n",
" \"relationships\": len(knowledge_graph.get(\"relationships\", [])),\n",
" \"conflicts_resolved\": len(resolved_conflicts),\n",
" \"merged_entities\": len(merged_entities),\n",
" \"communities\": num_communities,\n",
" \"entity_conflicts_resolved\": len(locals().get(\"resolved_entity_value_conflicts\", [])),\n",
" \"relationship_conflicts_resolved\": len(locals().get(\"resolved_relationship_conflicts\", [])),\n",
" \"deduplicated_entities\": len(locals().get(\"deduplicated_entities\", [])),\n",
" \"communities\": locals().get(\"num_communities\", 0),\n",
" \"questions_answered\": len(generated_answers),\n",
" \"llm_model\": groq_llm.model,\n",
" \"llm_model\": getattr(groq_llm, \"model\", \"unknown\"),\n",
"}\n",
"\n",
"print(\"Export completed\")\n",
"print(\"KG JSON entities:\", analysis_summary[\"entities\"])\n",
"print(\"KG RDF size (chars):\", len(kg_rdf))\n",
"print(\"KG RDF size (chars):\", len(kg_rdf_content))\n",
"print(\"Questions answered:\", analysis_summary[\"questions_answered\"])\n",
"print(\"LLM model:\", analysis_summary[\"llm_model\"])\n"
"print(\"LLM model:\", analysis_summary[\"llm_model\"])\n",
"print(\"Conflicts resolved:\", analysis_summary[\"entity_conflicts_resolved\"] + analysis_summary[\"relationship_conflicts_resolved\"])"
]
}
],
+5 -5
View File
@@ -12,22 +12,22 @@ How to cite Semantica in academic papers and research.
author = {Hawksight AI},
year = {2026},
url = {https://github.com/Hawksight-AI/semantica},
version = {0.2.2},
version = {0.2.3},
doi = {10.5281/zenodo.XXXXXXX}
}
```
### APA
Hawksight AI. (2026). *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering* (Version 0.2.2) [Computer software]. https://github.com/Hawksight-AI/semantica
Hawksight AI. (2026). *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering* (Version 0.2.3) [Computer software]. https://github.com/Hawksight-AI/semantica
### MLA
Hawksight AI. *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering*. Version 0.2.2, GitHub, 2026, https://github.com/Hawksight-AI/semantica.
Hawksight AI. *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering*. Version 0.2.3, GitHub, 2026, https://github.com/Hawksight-AI/semantica.
### Chicago
Hawksight AI. *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering*. Version 0.2.2. GitHub, 2026. https://github.com/Hawksight-AI/semantica.
Hawksight AI. *Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering*. Version 0.2.3. GitHub, 2026. https://github.com/Hawksight-AI/semantica.
### IEEE
Hawksight AI, "Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering," Version 0.2.2, GitHub, 2026. [Online]. Available: https://github.com/Hawksight-AI/semantica
Hawksight AI, "Semantica: An Open Source Framework for Semantic Layers and Knowledge Engineering," Version 0.2.3, GitHub, 2026. [Online]. Available: https://github.com/Hawksight-AI/semantica
---
+1 -1
View File
@@ -17,7 +17,7 @@
<p><em>The missing fabric between raw data and AI engineering. A comprehensive open-source framework for building semantic layers and knowledge engineering systems that transform unstructured data into AI-ready knowledge — powering Knowledge Graph-Powered RAG (GraphRAG), AI Agents, Multi-Agent Systems, and AI applications with structured semantic knowledge.</em></p>
<p>🆓 <strong>100% Open Source</strong> • 📜 <strong>MIT Licensed</strong> • 🚀 <strong>Latest Version: 0.1.1</strong> • 🚀 <strong>Production Ready</strong> • 🌍 <strong>Community Driven</strong></p>
<p>🆓 <strong>100% Open Source</strong> • 📜 <strong>MIT Licensed</strong> • 🚀 <strong>Latest Version: 0.2.3</strong> • 🚀 <strong>Production Ready</strong> • 🌍 <strong>Community Driven</strong></p>
<p>
<a href="getting-started/" class="md-button md-button--primary">Get Started</a>
+6 -1
View File
@@ -9,7 +9,12 @@ document.addEventListener("DOMContentLoaded", function () {
// Define versions
var versions = [
{ name: "0.1.1", url: "#", current: true },
{ name: "0.2.4", url: "#", current: true },
{ name: "0.2.3", url: "#", current: false },
{ name: "0.2.2", url: "#", current: false },
{ name: "0.2.1", url: "#", current: false },
{ name: "0.2.0", url: "#", current: false },
{ name: "0.1.1", url: "#", current: false },
{ name: "0.1.0", url: "#", current: false }
];
+34
View File
@@ -63,6 +63,29 @@ The module uses several inference algorithms:
---
## Ontology Ingestion
Ingest existing ontology files directly into usable data structures using `OntologyIngestor`.
**Function:** `ingest_ontology(source, method="file")`
| Argument | Description |
|----------|-------------|
| `source` | File path, directory path, or list of paths |
| `method` | Ingestion method (default: "file") |
**Example:**
```python
from semantica.ontology import ingest_ontology
# Ingest file
data = ingest_ontology("ontology.ttl")
# Ingest directory
dataset = ingest_ontology("ontologies/")
```
## Main Classes
### OntologyEngine
@@ -170,6 +193,17 @@ Manages external dependencies.
| `import_external_ontology(uri, ontology)` | Load and merge external ontology |
| `evaluate_alignment(uri, ontology)` | Assess alignment and compatibility |
### OntologyIngestor
Handles ingestion of existing ontologies from files and directories.
**Methods:**
| Method | Description |
|--------|-------------|
| `ingest_ontology(file_path)` | Ingest a single ontology file |
| `ingest_directory(directory_path)` | Recursively ingest ontology files from a directory |
---
## Unified Engine Examples
+14 -2
View File
@@ -112,6 +112,8 @@ The main facade for all vector operations.
| Method | Description |
|--------|-------------|
| `store_vectors(vectors, metadata)` | Store embeddings |
| `add_documents(documents, metadata, batch_size, parallel)` | **(New)** Store documents with automatic embedding generation and parallelization |
| `embed_batch(texts)` | **(New)** Generate embeddings for a batch of texts |
| `search(query, k)` | Semantic search |
| `delete(ids)` | Remove vectors |
@@ -120,15 +122,24 @@ The main facade for all vector operations.
```python
from semantica.vector_store import VectorStore
# Initialize (defaults to FAISS)
# Initialize (defaults to FAISS, parallel enabled by default with 6 workers)
store = VectorStore(backend="faiss", dimension=1536)
# Store
# 1. Store pre-computed vectors
ids = store.store_vectors(
vectors=[[0.1, 0.2, ...], ...],
metadata=[{"text": "Hello"}, ...]
)
# 2. Store raw documents (High Performance)
# Automatically handles embedding generation in parallel batches (uses default 6 workers)
ids = store.add_documents(
documents=["Doc 1", "Doc 2", ...],
metadata=[{"id": 1}, {"id": 2}, ...],
batch_size=32,
parallel=True
)
# Search
results = store.search(query_vector=[0.1, 0.2, ...], k=5)
```
@@ -771,6 +782,7 @@ print(f"Context: {context}")
---
## See Also
- [High-Performance Usage Guide](../vector_store_usage.md) - **(New)** Parallel ingestion and batching guide
- [Embeddings Module](embeddings.md) - Generates the vectors
- [Context Module](context.md) - Uses vector store for memory
- [Ingest Module](ingest.md) - Source of data
+101
View File
@@ -0,0 +1,101 @@
# High-Performance Vector Store Usage
This guide demonstrates how to leverage the new high-performance features of the Semantica Vector Store, specifically designed for efficient batch processing and parallel ingestion of large document sets.
## 🚀 Key Features
- **Parallel Ingestion**: Utilize multi-threading to embed and store documents concurrently.
- **Batch Processing**: Automatically group documents into batches to minimize overhead.
- **Unified API**: A single `add_documents` method handles embedding generation and storage.
---
## ⚡ Quick Start: Parallel Ingestion
The fastest way to ingest documents is using the `add_documents` method. Parallelization is enabled by default with optimized settings (6 workers).
```python
from semantica.vector_store import VectorStore
import time
store = VectorStore(
backend="faiss",
dimension=768,
)
documents = [f"This is document number {i} with some content." for i in range(1000)]
metadata = [{"source": "generated", "id": i} for i in range(1000)]
start_time = time.time()
ids = store.add_documents(
documents=documents,
metadata=metadata,
batch_size=64,
parallel=True,
)
print(f"Ingested {len(ids)} documents in {time.time() - start_time:.2f}s")
```
---
## 📊 Performance Comparison
### Old Method (Sequential Loop)
*Slower due to sequential processing and overhead per single item.*
```python
for doc in documents:
emb = embedder.generate(doc)
store.store_vectors([emb], [{"text": doc}])
```
### New Method (Parallel Batching)
*Significantly faster (3x-10x) by utilizing thread pools and batch operations.*
```python
store.add_documents(documents, parallel=True)
```
---
## 🛠 Configuration & Tuning
### `max_workers`
Controls the number of concurrent threads used for embedding generation.
- **Default**: 6 (Optimized for most systems)
- **Recommendation**: You generally don't need to change this. If you have very high core counts or specific throughput needs, you can override it.
```python
store = VectorStore(max_workers=16)
```
### `batch_size`
Controls how many documents are processed in a single chunk.
- **Default**: 32
- **Recommendation**:
- **Local Models**: 32-64 usually works well.
- **API Models (OpenAI, etc.)**: Larger batches (e.g., 100-200) can reduce network latency overhead.
```python
store.add_documents(documents, batch_size=100)
```
---
## 🧩 Advanced: Manual Batch Embedding
If you need the embeddings without storing them immediately, use `embed_batch`.
```python
vectors = store.embed_batch(
texts=documents[:100],
)
print(f"Generated {len(vectors)} vectors")
```
## ⚠️ Best Practices
1. **Metadata Consistency**: Ensure your `metadata` list has the same length as your `documents` list.
2. **Error Handling**: The `add_documents` method will propagate exceptions if embedding fails. Ensure your data is clean.
3. **Memory Usage**: Very large `batch_size` combined with high `max_workers` can increase memory usage. Monitor your system resources.
+1 -1
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "semantica"
version = "0.2.2"
version = "0.2.4"
description = "🧠 Semantica - An Open Source Framework for building Semantic Layers and Knowledge Engineering"
readme = "README.md"
license = { text = "MIT" }
+1 -1
View File
@@ -10,7 +10,7 @@ Main exports:
- Config: Configuration management
"""
__version__ = "0.2.2"
__version__ = "0.2.4"
__author__ = "Semantica Contributors"
__license__ = "MIT"
+9
View File
@@ -86,6 +86,7 @@ Main Classes:
- RepoIngestor: Git repository processing
- EmailIngestor: Email protocol handling
- DBIngestor: Database export handling
- OntologyIngestor: Ontology file processing
- MethodRegistry: Registry for custom ingestion methods
- IngestConfig: Configuration manager for ingest module
@@ -98,6 +99,7 @@ Convenience Functions:
- ingest_repository: Repository ingestion wrapper
- ingest_email: Email ingestion wrapper
- ingest_database: Database ingestion wrapper
- ingest_ontology: Ontology ingestion wrapper
Example Usage:
@@ -134,6 +136,7 @@ from .methods import (
ingest_feed,
ingest_file,
ingest_mcp,
ingest_ontology,
ingest_repository,
ingest_stream,
ingest_web,
@@ -166,6 +169,8 @@ from .web_ingestor import (
WebIngestor,
)
from .ontology_ingestor import OntologyData, OntologyIngestor
__all__ = [
# File ingestion
"FileIngestor",
@@ -216,6 +221,9 @@ __all__ = [
"MCPClient",
"MCPResource",
"MCPTool",
# Ontology ingestion
"OntologyIngestor",
"OntologyData",
# Registry and Methods
"MethodRegistry",
"method_registry",
@@ -227,6 +235,7 @@ __all__ = [
"ingest_repository",
"ingest_email",
"ingest_database",
"ingest_ontology",
"ingest_mcp",
"get_ingest_method",
"list_available_methods",
+69
View File
@@ -150,6 +150,7 @@ from .email_ingestor import EmailData, EmailIngestor
from .feed_ingestor import FeedData, FeedIngestor
from .file_ingestor import FileIngestor, FileObject
from .mcp_ingestor import MCPData, MCPIngestor
from .ontology_ingestor import OntologyData, OntologyIngestor
from .registry import method_registry
from .repo_ingestor import RepoIngestor
from .stream_ingestor import StreamIngestor, StreamProcessor
@@ -537,6 +538,66 @@ def ingest_email(
raise
def ingest_ontology(
source: Union[str, Path, List[Union[str, Path]]], method: str = "file", **kwargs
) -> Union[OntologyData, List[OntologyData]]:
"""
Ingest ontology from source (convenience function).
This is a user-friendly wrapper that ingests ontologies using the specified method.
Args:
source: Ontology file path, directory path, or list of paths
method: Ingestion method (default: "file")
- "file": Single file ingestion
- "directory": Directory ingestion with recursive scanning
**kwargs: Additional options passed to OntologyIngestor
Returns:
OntologyData, List[OntologyData] with ingestion results
Examples:
>>> from semantica.ingest.methods import ingest_ontology
>>> ontology = ingest_ontology("ontology.ttl")
>>> ontologies = ingest_ontology("./ontologies", method="directory")
"""
# Check for custom method in registry
custom_method = method_registry.get("ontology", method)
if custom_method and custom_method != ingest_ontology:
try:
return custom_method(source, **kwargs)
except Exception as e:
logger.warning(
f"Custom method {method} failed: {e}, falling back to default"
)
try:
# Get config
config = ingest_config.get_method_config("ontology")
config.update(kwargs)
ingestor = OntologyIngestor(**config)
source_path = str(source) if isinstance(source, (str, Path)) else None
if method == "file" and source_path:
if isinstance(source, list):
return [ingestor.ingest_ontology(str(s), **kwargs) for s in source]
return ingestor.ingest_ontology(source_path, **kwargs)
elif method == "directory" and source_path:
recursive = kwargs.get("recursive", ingest_config.get("recursive", True))
return ingestor.ingest_directory(source_path, recursive=recursive, **kwargs)
else:
# Default: try as file
if isinstance(source, list):
return [ingestor.ingest_ontology(str(s), **kwargs) for s in source]
return ingestor.ingest_ontology(str(source), **kwargs)
except Exception as e:
logger.error(f"Failed to ingest ontology: {e}")
raise
def ingest_database(
source: Union[str, Dict[str, Any]], method: Optional[str] = None, **kwargs
) -> Union[TableData, List[TableData], Dict[str, Any]]:
@@ -769,6 +830,7 @@ def ingest(
- "repo": Repository ingestion
- "email": Email ingestion
- "db": Database ingestion
- "ontology": Ontology ingestion
method: Optional specific ingestion method
**kwargs: Additional options passed to ingestor
@@ -802,6 +864,8 @@ def ingest(
("git@", "https://github.com", "https://gitlab.com")
):
source_type = "repo"
elif source_str.endswith((".ttl", ".owl", ".rdf", ".jsonld", ".n3", ".nt")):
source_type = "ontology"
else:
source_type = "file"
else:
@@ -830,6 +894,8 @@ def ingest(
raise ProcessingError("Email ingestion requires configuration dictionary")
elif source_type == "db":
return {"data": ingest_database(sources, method=method, **kwargs)}
elif source_type == "ontology":
return {"ontology": ingest_ontology(sources, method=method or "file", **kwargs)}
elif source_type == "mcp":
return {"data": ingest_mcp(sources, method=method or "resources", **kwargs)}
else:
@@ -909,5 +975,8 @@ method_registry.register("mcp", "default", ingest_mcp)
method_registry.register("mcp", "resources", ingest_mcp)
method_registry.register("mcp", "tools", ingest_mcp)
method_registry.register("mcp", "all", ingest_mcp)
method_registry.register("ontology", "default", ingest_ontology)
method_registry.register("ontology", "file", ingest_ontology)
method_registry.register("ontology", "directory", ingest_ontology)
method_registry.register("ingest", "default", ingest)
method_registry.register("ingest", "unified", ingest)
+392
View File
@@ -0,0 +1,392 @@
"""
Ontology Ingestion Module
This module provides capabilities to ingest external ontologies from files (OWL, RDF, TTL, etc.)
and convert them into Semantica's internal ontology dictionary format.
Supported Formats:
- Turtle (.ttl): Terse RDF Triple Language. A concise, human-readable
format for representing RDF graphs. Commonly used for writing
ontologies by hand.
- RDF/XML (.rdf, .owl): The XML serialization of RDF. The standard
format for OWL (Web Ontology Language) ontologies and often used
for data interchange.
- JSON-LD (.jsonld): JSON for Linked Data. A lightweight Linked Data
format that is easy for humans to read and for machines to parse
and generate. Ideal for web-based applications.
- N-Triples (.nt): A line-based, plain text format for encoding an
RDF graph. Each line represents a single triple. Very simple to
parse but verbose.
- Notation3 (.n3): A superset of Turtle that adds features like logic
and rules.
Key Features:
- Support for multiple RDF formats (Turtle, RDF/XML, JSON-LD, N3, NT)
- Automatic parsing using rdflib
- Conversion to Semantica ontology structure
- Batch processing of ontology files
- Extraction of classes, properties, and metadata
Example Usage:
>>> from semantica.ingest import OntologyIngestor
>>> ingestor = OntologyIngestor()
>>> ontology = ingestor.ingest_ontology("my_ontology.ttl")
"""
import os
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional, Union
import rdflib
from rdflib import RDF, RDFS, OWL, Graph
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
@dataclass
class OntologyData:
"""Ontology data representation."""
data: Dict[str, Any]
source_path: str
format: str
metadata: Dict[str, Any] = field(default_factory=dict)
ingested_at: datetime = field(default_factory=datetime.now)
class OntologyIngestor:
"""
Ontology ingestion handler.
This class parses OWL/RDF files and converts them to Semantica's ontology dictionary format.
"""
def __init__(self, config: Optional[Dict[str, Any]] = None, **kwargs):
"""
Initialize ontology ingestor.
Args:
config: Optional configuration dictionary
**kwargs: Additional configuration parameters
"""
self.logger = get_logger("ontology_ingestor")
self.progress = get_progress_tracker()
self.config = config or {}
self.config.update(kwargs)
def ingest_ontology(self, file_path: Union[str, Path], format: Optional[str] = None, **kwargs) -> OntologyData:
"""
Ingest an ontology file.
Args:
file_path: Path to the ontology file (string or Path object)
format: Optional format hint (e.g., 'turtle', 'xml'). If None, rdflib guesses.
**kwargs: Additional arguments for rdflib parsing
Returns:
OntologyData object containing the parsed ontology and metadata
"""
file_path = Path(file_path)
# Track file ingestion
tracking_id = self.progress.start_tracking(
file=str(file_path),
module="ingest",
submodule="OntologyIngestor",
message=f"Ontology: {file_path.name}",
)
try:
# Validate file exists
if not file_path.exists():
raise ValidationError(f"File not found: {file_path}")
self.progress.update_tracking(tracking_id, message="Parsing RDF graph...")
g = Graph()
# Use provided format or let rdflib guess based on extension
parse_kwargs = kwargs.copy()
if format:
parse_kwargs['format'] = format
try:
g.parse(file_path, **parse_kwargs)
except Exception as e:
# Fallback: try to guess format from extension if not provided and initial parse failed
if not format:
ext = os.path.splitext(file_path)[1].lower()
fmt_map = {
'.ttl': 'turtle',
'.owl': 'xml', # OWL is often XML
'.rdf': 'xml',
'.jsonld': 'json-ld',
'.n3': 'n3',
'.nt': 'nt'
}
guessed_fmt = fmt_map.get(ext)
if guessed_fmt:
self.logger.info(f"Retrying with guessed format: {guessed_fmt}")
g.parse(file_path, format=guessed_fmt, **kwargs)
else:
raise e
else:
raise e
self.progress.update_tracking(tracking_id, message="Converting to internal format...")
# Determine format for metadata
used_format = format
if not used_format:
ext = os.path.splitext(file_path)[1].lower()
fmt_map = {
'.ttl': 'turtle',
'.owl': 'xml',
'.rdf': 'xml',
'.jsonld': 'json-ld',
'.n3': 'n3',
'.nt': 'nt'
}
used_format = fmt_map.get(ext, 'unknown')
ontology_dict = self._convert_to_dict(g, source_path=str(file_path), format=used_format)
ontology_data = OntologyData(
data=ontology_dict,
source_path=str(file_path),
format=used_format,
metadata=ontology_dict.get("metadata", {}).copy()
)
self.progress.stop_tracking(
tracking_id,
status="completed",
message=f"Successfully ingested ontology from {file_path}",
)
return ontology_data
except Exception as e:
self.logger.error(f"Failed to ingest ontology: {str(e)}")
self.progress.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise ProcessingError(f"Failed to ingest ontology: {str(e)}") from e
def ingest_directory(self, directory_path: Union[str, Path], recursive: bool = True, **kwargs) -> List[OntologyData]:
"""
Ingest all ontology files in a directory.
Args:
directory_path: Path to the directory (string or Path object)
recursive: Whether to search recursively
**kwargs: Additional arguments
Returns:
List of OntologyData objects
"""
directory_path = Path(directory_path)
ontologies = []
extensions = {'.ttl', '.owl', '.rdf', '.jsonld', '.n3', '.nt'}
# Track directory ingestion
tracking_id = self.progress.start_tracking(
file=str(directory_path),
module="ingest",
submodule="OntologyIngestor",
message=f"Directory: {directory_path.name}",
)
try:
if not directory_path.exists():
raise ValidationError(f"Directory not found: {directory_path}")
if not directory_path.is_dir():
raise ValidationError(f"Path is not a directory: {directory_path}")
files_to_process = []
for root, _, files in os.walk(directory_path):
for file in files:
ext = os.path.splitext(file)[1].lower()
if ext in extensions:
files_to_process.append(os.path.join(root, file))
if not recursive:
break
total_files = len(files_to_process)
self.progress.update_tracking(
tracking_id, message=f"Processing {total_files} ontology files"
)
for idx, file_path in enumerate(files_to_process, 1):
try:
ont_data = self.ingest_ontology(file_path, **kwargs)
ontologies.append(ont_data)
self.progress.update_progress(
tracking_id,
processed=idx,
total=total_files,
message=f"Processing {idx}/{total_files}: {Path(file_path).name}"
)
except Exception as e:
self.logger.warning(f"Skipping {file_path}: {e}")
self.progress.stop_tracking(
tracking_id,
status="completed",
message=f"Ingested {len(ontologies)} ontologies",
)
return ontologies
except Exception as e:
self.progress.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def _convert_to_dict(self, graph: Graph, source_path: str, format: str = "unknown") -> Dict[str, Any]:
"""
Convert rdflib Graph to Semantica ontology dictionary.
Args:
graph: Parsed rdflib Graph
source_path: Source file path
format: Format of the ontology file
Returns:
Ontology dictionary
"""
ontology = {
"uri": "",
"name": os.path.basename(source_path),
"version": "1.0",
"classes": [],
"properties": [],
"metadata": {
"source_path": source_path,
"ingested_at": datetime.now().isoformat(),
"format": format
}
}
# 1. Extract Ontology Metadata
for s, p, o in graph.triples((None, RDF.type, OWL.Ontology)):
ontology["uri"] = str(s)
# Try to find label/comment/versionInfo
for _, _, label in graph.triples((s, RDFS.label, None)):
ontology["name"] = str(label)
for _, _, comment in graph.triples((s, RDFS.comment, None)):
ontology["description"] = str(comment)
for _, _, version in graph.triples((s, OWL.versionInfo, None)):
ontology["version"] = str(version)
# Break after first ontology definition found (usually only one per file)
break
# 2. Extract Classes
classes = {}
# Union of owl:Class and rdfs:Class
class_types = [OWL.Class, RDFS.Class]
for c_type in class_types:
for s, p, o in graph.triples((None, RDF.type, c_type)):
if isinstance(s, rdflib.BNode):
continue # Skip blank nodes for now
uri = str(s)
if uri not in classes:
cls_def = {
"uri": uri,
"name": self._get_local_name(uri),
"type": "class"
}
# Add label/comment
label = graph.value(s, RDFS.label)
if label:
cls_def["label"] = str(label)
cls_def["name"] = str(label) # Prefer label as name if available? Or keep URI fragment?
# Keeping local name from URI is safer for internal IDs, label for display.
# But Semantica seems to use "name" for the identifier in some examples.
# Let's keep name as local name or label if simple.
comment = graph.value(s, RDFS.comment)
if comment:
cls_def["description"] = str(comment)
# Superclasses
parents = []
for _, _, parent in graph.triples((s, RDFS.subClassOf, None)):
if isinstance(parent, rdflib.URIRef):
parents.append(str(parent))
if parents:
cls_def["parents"] = parents
classes[uri] = cls_def
ontology["classes"] = list(classes.values())
# 3. Extract Properties
properties = {}
# Object Properties
for s, p, o in graph.triples((None, RDF.type, OWL.ObjectProperty)):
self._add_property(graph, s, "object", properties)
# Datatype Properties
for s, p, o in graph.triples((None, RDF.type, OWL.DatatypeProperty)):
self._add_property(graph, s, "data", properties)
# RDF Properties (generic)
for s, p, o in graph.triples((None, RDF.type, RDF.Property)):
if str(s) not in properties: # Don't overwrite if already found as specific type
self._add_property(graph, s, "annotation", properties) # Default to annotation or generic
ontology["properties"] = list(properties.values())
return ontology
def _add_property(self, graph: Graph, subject: rdflib.term.Node, prop_type: str, properties_dict: Dict):
if isinstance(subject, rdflib.BNode):
return
uri = str(subject)
if uri in properties_dict:
return
prop_def = {
"uri": uri,
"name": self._get_local_name(uri),
"type": prop_type
}
label = graph.value(subject, RDFS.label)
if label:
prop_def["label"] = str(label)
comment = graph.value(subject, RDFS.comment)
if comment:
prop_def["description"] = str(comment)
# Domain and Range
domain = graph.value(subject, RDFS.domain)
if domain and isinstance(domain, rdflib.URIRef):
prop_def["domain"] = str(domain)
range_val = graph.value(subject, RDFS.range)
if range_val and isinstance(range_val, rdflib.URIRef):
prop_def["range"] = str(range_val)
properties_dict[uri] = prop_def
def _get_local_name(self, uri: str) -> str:
"""Extract local name from URI."""
if '#' in uri:
return uri.split('#')[-1]
return uri.split('/')[-1]
+51 -5
View File
@@ -161,7 +161,14 @@ class GraphBuilder:
}
all_relationships.append(rel_dict)
elif isinstance(item, dict):
# Detect and normalize Entity objects inside dict
if "source_id" in item and "source" not in item:
item["source"] = item["source_id"]
if "target_id" in item and "target" not in item:
item["target"] = item["target_id"]
if "subject" in item and "source" not in item:
item["source"] = item["subject"]
if "object" in item and "target" not in item:
item["target"] = item["object"]
if "source" in item and not isinstance(item["source"], str):
src = item["source"]
item["source"] = getattr(src, "id", getattr(src, "text", str(src)))
@@ -347,6 +354,21 @@ class GraphBuilder:
elif not isinstance(sources, list):
sources = [sources]
# Count input relationships for warning if all are dropped
input_relationships_count = 0
if isinstance(source_dict, dict):
rels = source_dict.get("relationships", [])
if isinstance(rels, list):
input_relationships_count += len(rels)
elif rels is not None:
input_relationships_count += 1
if explicit_relationships:
for rel_item in explicit_relationships:
if isinstance(rel_item, list):
input_relationships_count += len(rel_item)
else:
input_relationships_count += 1
# Track graph building
build_start_time = time.time()
@@ -468,11 +490,12 @@ class GraphBuilder:
pipeline_id=pipeline_id,
)
# Check if relationships are already in dictionary format
sample_rel = relationships_list[0] if relationships_list else None
is_dict_format = isinstance(sample_rel, dict) and (
"source" in sample_rel and "target" in sample_rel
) and not hasattr(sample_rel, "__dict__") # Ensure it's not a class instance
("source" in sample_rel and "target" in sample_rel)
or ("source_id" in sample_rel and "target_id" in sample_rel)
or ("subject" in sample_rel and "object" in sample_rel)
) and not hasattr(sample_rel, "__dict__")
if is_dict_format:
# Fast path: directly append dictionaries after normalizing source/target
@@ -481,8 +504,15 @@ class GraphBuilder:
batch = relationships_list[i:i + batch_size]
for item in batch:
if isinstance(item, dict):
# Normalize source/target if they are objects
rel_dict = item.copy()
if "source_id" in rel_dict and "source" not in rel_dict:
rel_dict["source"] = rel_dict["source_id"]
if "target_id" in rel_dict and "target" not in rel_dict:
rel_dict["target"] = rel_dict["target_id"]
if "subject" in rel_dict and "source" not in rel_dict:
rel_dict["source"] = rel_dict["subject"]
if "object" in rel_dict and "target" not in rel_dict:
rel_dict["target"] = rel_dict["object"]
if "source" in rel_dict and not isinstance(rel_dict["source"], str):
src = rel_dict["source"]
rel_dict["source"] = getattr(src, "id", getattr(src, "text", str(src)))
@@ -571,6 +601,14 @@ class GraphBuilder:
f"Entity resolution complete: {len(all_entities)} -> {len(resolved_entities)} unique entities"
)
if input_relationships_count > 0 and len(all_relationships) == 0:
warning_msg = (
f"All relationships were dropped during graph building: "
f"{input_relationships_count} input relationships, 0 in final graph"
)
self.logger.warning(warning_msg)
print(f"Warning: {warning_msg}")
# Build graph structure
print("Building graph structure...")
structure_start = time.time()
@@ -682,6 +720,14 @@ class GraphBuilder:
)
raise
def build_single_source(
self,
kg_data: Dict[str, Any],
pipeline_id: Optional[str] = None,
**options,
) -> Dict[str, Any]:
return self.build(kg_data, pipeline_id=pipeline_id, **options)
def add_temporal_edge(
self,
graph,
+8 -1
View File
@@ -109,12 +109,14 @@ Convenience Functions:
- create_associative_class: Associative class creation wrapper
- get_ontology_method: Get ontology method by name
- list_available_methods: List registered methods
- ingest_ontology: Ingest ontology from file or directory
Example Usage:
>>> from semantica.ontology import generate_ontology, infer_classes, OntologyGenerator
>>> from semantica.ontology import generate_ontology, infer_classes, OntologyGenerator, ingest_ontology
>>> # Using convenience functions
>>> ontology = generate_ontology({"entities": [...], "relationships": [...]}, method="default")
>>> classes = infer_classes(entities, method="default")
>>> data = ingest_ontology("ontology.ttl")
>>> # Using classes directly
>>> from semantica.ontology import OntologyGenerator, ClassInferrer, PropertyGenerator
>>> generator = OntologyGenerator(base_uri="https://example.org/ontology/")
@@ -155,6 +157,8 @@ from .registry import MethodRegistry, method_registry
from .requirements_spec import RequirementsSpec, RequirementsSpecManager
from .reuse_manager import ReuseDecision, ReuseManager
from .version_manager import OntologyVersion, VersionManager
from semantica.ingest import OntologyData, OntologyIngestor
from .methods import ingest_ontology
__all__ = [
# Main generators
@@ -200,4 +204,7 @@ __all__ = [
# Configuration
"OntologyConfig",
"ontology_config",
"ingest_ontology",
"OntologyData",
"OntologyIngestor",
]
+26 -3
View File
@@ -111,15 +111,19 @@ Main Functions:
- create_associative_class: Associative class creation wrapper
- get_ontology_method: Get ontology method by name
- list_available_methods: List registered methods
- ingest_ontology: Ingest ontology from file or directory (via semantica.ingest)
Example Usage:
>>> from semantica.ontology.methods import generate_ontology, infer_classes
>>> from semantica.ontology.methods import generate_ontology, infer_classes, ingest_ontology
>>> ontology = generate_ontology({"entities": [...], "relationships": [...]}, method="default")
>>> classes = infer_classes(entities, method="default")
>>> data = ingest_ontology("ontology.ttl")
"""
from typing import Any, Callable, Dict, List, Optional
from typing import Any, Callable, Dict, List, Optional, Union
from pathlib import Path
from semantica.ingest import ingest_ontology as _ingest_ontology, OntologyData
from .registry import method_registry
@@ -172,4 +176,23 @@ def list_available_methods(task: Optional[str] = None) -> Dict[str, List[str]]:
return method_registry.list_all(task)
pass
def ingest_ontology(
source: Union[str, Path, List[Union[str, Path]]],
method: str = "file",
**kwargs
) -> Union[OntologyData, List[OntologyData]]:
"""
Ingest ontology from source.
This is a convenience wrapper around semantica.ingest.ingest_ontology.
Args:
source: Ontology file path, directory path, or list of paths
method: Ingestion method (default: "file")
**kwargs: Additional options
Returns:
OntologyData or List[OntologyData]
"""
return _ingest_ontology(source, method=method, **kwargs)
+55
View File
@@ -42,6 +42,18 @@ classes = inferrer.infer_classes(entities, build_hierarchy=True)
properties = prop_gen.infer_properties(entities, relationships, classes)
```
### Ingesting Ontologies
```python
from semantica.ingest import OntologyIngestor
# Create ingestor
ingestor = OntologyIngestor()
# Ingest ontology
ontology_data = ingestor.ingest_ontology("ontology.ttl")
```
## Ontology Generation
### Basic Ontology Generation
@@ -115,6 +127,49 @@ ontology = engine.from_data(
)
```
## Ontology Ingestion
### Basic Ingestion
Ingest existing ontologies from files (Turtle, RDF/XML, JSON-LD, etc.) into `OntologyData` objects.
```python
from semantica.ontology import ingest_ontology
# Ingest a single file
ontology_data = ingest_ontology("path/to/ontology.ttl")
print(f"Source: {ontology_data.source_path}")
print(f"Format: {ontology_data.format}")
print(f"Data keys: {ontology_data.data.keys()}")
```
### Ingesting Directories
Ingest all ontology files in a directory recursively.
```python
from semantica.ontology import ingest_ontology
# Ingest a directory
ontologies = ingest_ontology("path/to/ontologies_dir/")
for ont in ontologies:
print(f"Ingested: {ont.source_path} ({ont.format})")
```
### Unified Ingestion Interface
You can also use the unified `semantica.ingest` interface.
```python
from semantica.ingest import ingest
# Ingest as "ontology" source type
result = ingest("path/to/ontology.ttl", source_type="ontology")
ontology_data = result["ontology"]
```
## Class Inference
### Basic Class Inference
+87
View File
@@ -363,3 +363,90 @@ class ReuseManager:
def list_known_ontologies(self) -> List[str]:
"""List known ontology URIs."""
return list(self.known_ontologies.keys())
def merge_ontology_data(
self, target: Dict[str, Any], source: Dict[str, Any], **options
) -> Dict[str, Any]:
"""
Merge source ontology data into target ontology.
Merges classes, properties, and metadata from source to target.
Handles deduplication based on URI and name.
Args:
target: Target ontology dictionary (modified in-place)
source: Source ontology dictionary
**options: Merge options:
- overwrite: Whether to overwrite existing elements (default: False)
- merge_metadata: Whether to merge metadata (default: True)
Returns:
Merged target ontology
"""
tracking_id = self.progress_tracker.start_tracking(
module="ontology",
submodule="ReuseManager",
message=f"Merging ontology {source.get('name', 'unknown')} into {target.get('name', 'unknown')}",
)
try:
overwrite = options.get("overwrite", False)
# Helper to merge lists of dicts (classes/properties)
def merge_lists(target_list, source_list, key_field="uri"):
existing_keys = {item.get(key_field): i for i, item in enumerate(target_list) if item.get(key_field)}
for item in source_list:
key = item.get(key_field)
if not key:
# Fallback to name if URI missing
key = item.get("name")
if key in existing_keys:
if overwrite:
target_list[existing_keys[key]] = item
else:
target_list.append(item)
if key:
existing_keys[key] = len(target_list) - 1
# Merge Classes
if "classes" in source:
if "classes" not in target:
target["classes"] = []
merge_lists(target["classes"], source["classes"])
# Merge Properties
if "properties" in source:
if "properties" not in target:
target["properties"] = []
merge_lists(target["properties"], source["properties"])
# Merge Metadata
if options.get("merge_metadata", True) and "metadata" in source:
if "metadata" not in target:
target["metadata"] = {}
# Update with source metadata, preserving target's specific fields if needed
# Here we just update
target["metadata"].update(source["metadata"])
# Merge Imports
if "imports" in source:
if "imports" not in target:
target["imports"] = []
for imp in source["imports"]:
if imp not in target["imports"]:
target["imports"].append(imp)
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Merged ontology data successfully",
)
return target
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
+6 -2
View File
@@ -32,12 +32,14 @@ License: MIT
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Union
from typing import Any, Callable, Dict, List, Optional, Union, TYPE_CHECKING
from ..utils.exceptions import ProcessingError, ValidationError
from ..utils.logging import get_logger
from ..utils.progress_tracker import get_progress_tracker
from .pipeline_validator import PipelineValidator
if TYPE_CHECKING:
from .pipeline_validator import PipelineValidator
class StepStatus(Enum):
@@ -104,6 +106,8 @@ class PipelineBuilder:
if not self.progress_tracker.enabled:
self.progress_tracker.enabled = True
from .pipeline_validator import PipelineValidator
self.validator = PipelineValidator(**self.config)
self.steps: List[PipelineStep] = []
self.step_registry: Dict[str, Callable] = {}
@@ -240,6 +240,12 @@ class CoreferenceResolver:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
verbose_mode = options.get("verbose", False) or self.config.get("verbose", False)
if verbose_mode:
import sys
print(f" [CoreferenceResolver] ERROR: Resolution failed: {e}", flush=True, file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
raise
def resolve(
+6 -1
View File
@@ -108,7 +108,12 @@ class LLMExtraction:
# Initialize provider using new system
try:
self.provider = create_provider(provider, **config)
# Sanitize config: remove api_key if it's None/empty to allow fallback
provider_config = config.copy()
if "api_key" in provider_config and not provider_config["api_key"]:
del provider_config["api_key"]
self.provider = create_provider(provider, **provider_config)
except Exception as e:
self.logger.warning(f"Failed to initialize {provider} provider: {e}")
self.provider = None
+83 -32
View File
@@ -786,7 +786,7 @@ def extract_entities_llm(
# Pass api_key if provided in kwargs (needed for all providers)
provider_kwargs = kwargs.copy()
if "api_key" not in provider_kwargs:
if "api_key" not in provider_kwargs or not provider_kwargs["api_key"]:
# Try to get from environment as fallback for all providers
import os
env_key = f"{provider.upper()}_API_KEY"
@@ -1563,18 +1563,24 @@ def extract_relations_llm(
# Pass api_key if provided in kwargs
provider_kwargs = kwargs.copy()
if "api_key" not in provider_kwargs:
# Check if api_key is provided but empty, or not provided at all
if "api_key" not in provider_kwargs or not provider_kwargs["api_key"]:
import os
env_key = f"{provider.upper()}_API_KEY"
api_key = os.getenv(env_key)
if api_key:
provider_kwargs["api_key"] = api_key
# Remove None/empty API key if still present to avoid provider errors
if "api_key" in provider_kwargs and not provider_kwargs["api_key"]:
del provider_kwargs["api_key"]
# 2. PROVIDER VALIDATION
try:
llm = create_provider(provider, model=model, **provider_kwargs)
if not llm.is_available():
error_msg = f"{provider} provider not available for relation extraction."
error_msg = f"{provider} provider not available for relation extraction (key missing?)."
logger.error(error_msg)
if not silent_fail:
raise ProcessingError(error_msg)
@@ -1606,12 +1612,8 @@ def extract_relations_llm(
)
original_entities = entities
max_entities_prompt = kwargs.get("max_entities_prompt", kwargs.get("max_entities", 80))
try:
max_entities_prompt = int(max_entities_prompt)
except Exception:
max_entities_prompt = 80
# Use a fixed internal default for prompt entity cap (do not accept overrides from kwargs)
max_entities_prompt = 80
prompt_entities = original_entities
if max_entities_prompt > 0 and len(original_entities) > max_entities_prompt:
prompt_entities = filter_entities_for_text(
@@ -1635,6 +1637,11 @@ If a relation doesn't fit any of the preferred types, use the most appropriate t
Extract meaningful relationships between entities. Use appropriate relation types that accurately describe how entities are connected.
Common relation types include: related_to, part_of, located_in, created_by, uses, depends_on, interacts_with, and similar variations."""
verbose_mode = kwargs.get("verbose", False)
if verbose_mode:
import sys
print(f" [methods.extract_relations_llm] Constructing prompt for {len(prompt_entities)} entities...", flush=True, file=sys.stdout)
if not SCHEMAS_AVAILABLE:
raise ImportError("Pydantic schemas not available. Install pydantic/instructor to use LLM extraction.")
@@ -1663,29 +1670,65 @@ Entities found in text: {entities_str}"""
try:
# Use typed generation with Pydantic schema
# Pass kwargs to allow max_tokens and other parameters to be used
result_obj = llm.generate_typed(prompt, schema=RelationsResponse, **kwargs)
if verbose_mode:
import sys
print(f" [methods.extract_relations_llm] Calling llm.generate_typed ({provider}/{model})...", flush=True, file=sys.stdout)
# Only forward minimal, safe parameters to provider calls
call_kwargs = {}
if "temperature" in kwargs:
call_kwargs["temperature"] = kwargs["temperature"]
if "verbose" in kwargs:
call_kwargs["verbose"] = kwargs["verbose"]
result_obj = llm.generate_typed(prompt, schema=RelationsResponse, **call_kwargs)
if verbose_mode:
import sys
print(f" [methods.extract_relations_llm] Received response from {provider}.", flush=True, file=sys.stdout)
# Convert back to internal Relation format
relations = []
for r_out in result_obj.relations:
# Find matching entities using hybrid similarity
subject_entity = match_entity(r_out.subject, original_entities)
object_entity = match_entity(r_out.object, original_entities)
if subject_entity and object_entity:
relations.append(Relation(
subject=subject_entity,
predicate=r_out.predicate,
object=object_entity,
confidence=r_out.confidence,
context=text, # Simplified context
metadata={
"provider": provider,
"model": model,
"extraction_method": "llm_typed"
}
))
# Convert back to internal Relation format (robust across providers)
# Normalize typed result to a plain dict compatible with _parse_relation_result
try:
if hasattr(result_obj, "model_dump"):
parsed = result_obj.model_dump()
elif isinstance(result_obj, dict):
parsed = result_obj
elif hasattr(result_obj, "relations"):
# Instructor may return objects for each relation; convert where possible
rel_items = []
for r in getattr(result_obj, "relations", []):
if hasattr(r, "model_dump"):
rel_items.append(r.model_dump())
elif isinstance(r, dict):
rel_items.append(r)
else:
# Best-effort attribute access
rel_items.append({
"subject": getattr(r, "subject", ""),
"object": getattr(r, "object", ""),
"predicate": getattr(r, "predicate", "related_to"),
"confidence": getattr(r, "confidence", 0.9),
})
parsed = {"relations": rel_items}
else:
parsed = result_obj
except Exception:
parsed = result_obj
# Use common parser to build internal Relation objects
relations = _parse_relation_result(parsed, original_entities, text, provider, model)
# If typed path returned no relations, attempt a structured JSON fallback
if not relations:
try:
if verbose_mode:
import sys
print(" [methods.extract_relations_llm] Typed result empty, attempting structured JSON fallback...", flush=True, file=sys.stdout)
raw_json = llm.generate_structured(prompt, **call_kwargs)
relations = _parse_relation_result(raw_json, original_entities, text, provider, model)
except Exception as _e:
# Keep relations as empty if fallback fails
pass
logger.info(f"Successfully extracted {len(relations)} relations using {provider}/{model} (typed)")
_result_cache.set("relations", text, relations, **cache_params)
return relations
@@ -1813,6 +1856,8 @@ def _extract_relations_chunked(
logger.debug(f"Scheduling relation extraction for chunk {i+1}/{len(chunks)} with {len(chunk_entities)} entities")
# Only pass minimal kwargs downstream
limited_kwargs = {k: kwargs[k] for k in ("relation_types", "temperature", "verbose") if k in kwargs}
future = executor.submit(
extract_relations_llm,
chunk.text,
@@ -1822,7 +1867,7 @@ def _extract_relations_chunked(
silent_fail=False,
max_text_length=len(chunk.text) + 1,
structured_output_mode=structured_output_mode,
**kwargs
**limited_kwargs
)
future_to_chunk[future] = i
@@ -2009,12 +2054,18 @@ def extract_triplets_llm(
# Pass api_key if provided in kwargs
provider_kwargs = kwargs.copy()
if "api_key" not in provider_kwargs:
# Check if api_key is provided but empty, or not provided at all
if "api_key" not in provider_kwargs or not provider_kwargs["api_key"]:
import os
env_key = f"{provider.upper()}_API_KEY"
api_key = os.getenv(env_key)
if api_key:
provider_kwargs["api_key"] = api_key
# Remove None/empty API key if still present to avoid provider errors
if "api_key" in provider_kwargs and not provider_kwargs["api_key"]:
del provider_kwargs["api_key"]
# 2. PROVIDER VALIDATION
try:
+6 -7
View File
@@ -375,14 +375,13 @@ class NERExtractor:
method_options["model"] = all_options.get(
"llm_model", all_options.get("model")
)
# Pass api_key if provided (needed for all providers)
if "api_key" in all_options:
method_options["api_key"] = all_options["api_key"]
elif "api_key" not in method_options:
# Try to get from environment as fallback
# Ensure api_key is populated: check explicitly provided or fallback to env
current_key = method_options.get("api_key")
if not current_key:
# Not found or empty/None, try environment
import os
provider = method_options.get("provider", "openai")
env_key = f"{provider.upper()}_API_KEY"
provider_name = method_options.get("provider", "openai")
env_key = f"{provider_name.upper()}_API_KEY"
api_key = os.getenv(env_key)
if api_key:
method_options["api_key"] = api_key
+149 -57
View File
@@ -243,68 +243,136 @@ class BaseProvider:
mode = instructor.Mode.TOOLS # Default mode
if provider_name == "OpenAIProvider" and self.client:
client = instructor.from_openai(self.client)
elif provider_name == "AnthropicProvider" and self.client:
client = instructor.from_anthropic(self.client)
elif provider_name == "GeminiProvider" and self.client:
client = instructor.from_gemini(
self.client,
mode=instructor.Mode.GEMINI_JSON
)
elif provider_name == "GroqProvider" and self.client:
# Try using from_groq if available (newer instructor versions)
if hasattr(instructor, "from_groq"):
client = instructor.from_groq(self.client, mode=instructor.Mode.JSON)
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"openai/{kwargs.get('model', self.model)}",
api_key=self.api_key
)
except Exception:
client = instructor.from_openai(self.client)
else:
# Fallback: Create OpenAI client pointing to Groq
# This avoids the "Client should be an instance of openai.OpenAI" warning
client = instructor.from_openai(self.client)
elif provider_name == "AnthropicProvider" and self.client:
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"anthropic/{kwargs.get('model', self.model)}",
api_key=self.api_key
)
except Exception:
client = instructor.from_anthropic(self.client)
else:
client = instructor.from_anthropic(self.client)
elif provider_name == "GeminiProvider" and self.client:
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"gemini/{kwargs.get('model', self.model)}",
api_key=self.api_key
)
except Exception:
client = instructor.from_gemini(
self.client,
mode=instructor.Mode.GEMINI_JSON
)
else:
client = instructor.from_gemini(
self.client,
mode=instructor.Mode.GEMINI_JSON
)
elif provider_name == "GroqProvider" and self.client:
# Try using from_provider which is recommended for Groq in latest instructor
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"groq/{kwargs.get('model', self.model)}",
api_key=self.api_key
)
except Exception:
client = None
if not client:
# Try using from_groq if available (newer instructor versions)
if hasattr(instructor, "from_groq"):
client = instructor.from_groq(self.client, mode=instructor.Mode.JSON)
else:
# Fallback: Create OpenAI client pointing to Groq
# This avoids the "Client should be an instance of openai.OpenAI" warning
try:
from openai import OpenAI
# Fix: Use self.api_key instead of self.client.api_key
groq_client = OpenAI(
base_url="https://api.groq.com/openai/v1",
api_key=self.api_key,
)
client = instructor.from_openai(groq_client, mode=instructor.Mode.JSON)
except Exception:
# Last resort: try passing the groq client directly
client = instructor.from_openai(self.client, mode=instructor.Mode.JSON)
elif provider_name == "OllamaProvider":
# Try from_provider for Ollama if available
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"ollama/{kwargs.get('model', self.model)}",
)
except Exception:
client = None
if not client:
# Create OpenAI-compatible client for Ollama
try:
from openai import OpenAI
groq_client = OpenAI(
base_url="https://api.groq.com/openai/v1",
api_key=self.client.api_key,
# Ollama typically runs on localhost:11434/v1
base_url = getattr(self, "base_url", "http://localhost:11434")
if not base_url.endswith("/v1"):
base_url = f"{base_url.rstrip('/')}/v1"
ollama_client = OpenAI(
base_url=base_url,
api_key="ollama", # required but unused
)
client = instructor.from_openai(groq_client, mode=instructor.Mode.JSON)
except Exception:
# Last resort: try passing the groq client directly
client = instructor.from_openai(self.client, mode=instructor.Mode.JSON)
elif provider_name == "OllamaProvider":
# Create OpenAI-compatible client for Ollama
try:
from openai import OpenAI
# Ollama typically runs on localhost:11434/v1
base_url = getattr(self, "base_url", "http://localhost:11434")
if not base_url.endswith("/v1"):
base_url = f"{base_url.rstrip('/')}/v1"
ollama_client = OpenAI(
base_url=base_url,
api_key="ollama", # required but unused
)
client = instructor.from_openai(ollama_client, mode=instructor.Mode.JSON)
except ImportError:
pass
client = instructor.from_openai(ollama_client, mode=instructor.Mode.JSON)
except ImportError:
pass
elif provider_name == "DeepSeekProvider" and self.client:
# DeepSeek is OpenAI compatible
# We need to wrap the underlying client if it exposes the OpenAI interface
# or create a new OpenAI client if self.client is a deepseek.Client (which might be just a wrapper)
# Assuming deepseek.Client is compatible or we can use OpenAI client
try:
# DeepSeek usually works with standard OpenAI client
# If self.client is deepseek.Client, check if we can wrap it
# Otherwise create a new OpenAI client
from openai import OpenAI
if isinstance(self.client, OpenAI):
client = instructor.from_openai(self.client, mode=instructor.Mode.JSON)
else:
# Try creating fresh client
ds_client = OpenAI(
api_key=self.api_key,
base_url="https://api.deepseek.com"
)
client = instructor.from_openai(ds_client, mode=instructor.Mode.JSON)
except Exception:
pass
# Try from_provider for DeepSeek
if hasattr(instructor, "from_provider"):
try:
client = instructor.from_provider(
provider=f"deepseek/{kwargs.get('model', self.model)}",
api_key=self.api_key
)
except Exception:
client = None
if not client:
# DeepSeek is OpenAI compatible
try:
from openai import OpenAI
if isinstance(self.client, OpenAI):
client = instructor.from_openai(self.client, mode=instructor.Mode.JSON)
else:
# Try creating fresh client
ds_client = OpenAI(
api_key=self.api_key,
base_url="https://api.deepseek.com"
)
client = instructor.from_openai(ds_client, mode=instructor.Mode.JSON)
except Exception:
pass
# Global LiteLLM support - if litellm is passed in kwargs or config
if not client and (kwargs.get("litellm") or self.config.get("litellm")):
if hasattr(instructor, "from_provider"):
try:
# Format for litellm in instructor is litellm/model_name
provider_model = kwargs.get("model", self.model)
litellm_provider = f"litellm/{provider_model}"
client = instructor.from_provider(litellm_provider, api_key=self.api_key)
except Exception:
pass
if client:
# Map generate arguments to client arguments
@@ -318,6 +386,11 @@ class BaseProvider:
"temperature": kwargs.get("temperature", 0.1), # Low temp for structured
}
verbose_mode = kwargs.get("verbose", False)
if verbose_mode:
import sys
print(f" [BaseProvider.generate_typed] Using instructor via {provider_name}. Client: {type(client)}", flush=True, file=sys.stdout)
# Pass through other common parameters
for param in ["max_tokens", "max_completion_tokens", "top_p", "frequency_penalty", "presence_penalty", "seed", "stop", "logit_bias", "user", "top_k"]:
if param in kwargs:
@@ -328,6 +401,9 @@ class BaseProvider:
create_kwargs["response_format"] = {"type": "json_object"}
response = client.chat.completions.create(**create_kwargs)
if verbose_mode:
import sys
print(f" [BaseProvider.generate_typed] Typed response received via instructor ({provider_name}).", flush=True, file=sys.stdout)
return response
except Exception as e:
self.logger.warning(f"Instructor generation failed ({e}), falling back to manual repair loop.")
@@ -706,7 +782,15 @@ class GroqProvider(BaseProvider):
if param in kwargs:
create_kwargs[param] = kwargs[param]
verbose_mode = kwargs.get("verbose", False)
if verbose_mode:
import sys
print(f" [GroqProvider.generate] Sending request to Groq API (model: {create_kwargs['model']})...", flush=True, file=sys.stdout)
response = self.client.chat.completions.create(**create_kwargs)
if verbose_mode:
import sys
print(f" [GroqProvider.generate] Response received from Groq.", flush=True, file=sys.stdout)
return response.choices[0].message.content
def generate_structured(self, prompt: str, **kwargs) -> dict:
@@ -737,7 +821,15 @@ class GroqProvider(BaseProvider):
if param in kwargs:
create_kwargs[param] = kwargs[param]
verbose_mode = kwargs.get("verbose", False)
if verbose_mode:
import sys
print(f" [GroqProvider.generate_structured] Sending structured request to Groq API (model: {create_kwargs['model']})...", flush=True, file=sys.stdout)
response = self.client.chat.completions.create(**create_kwargs)
if verbose_mode:
import sys
print(f" [GroqProvider.generate_structured] Structured response received from Groq.", flush=True, file=sys.stdout)
try:
return self._parse_json(response.choices[0].message.content)
except Exception as e:
@@ -419,14 +419,13 @@ class RelationExtractor:
method_options["model"] = all_options.get(
"llm_model", all_options.get("model")
)
# Pass api_key if provided (needed for all providers)
if "api_key" in all_options:
method_options["api_key"] = all_options["api_key"]
elif "api_key" not in method_options:
# Try to get from environment as fallback
# Ensure api_key is populated: check explicitly provided or fallback to env
current_key = method_options.get("api_key")
if not current_key:
# Not found or empty/None, try environment
import os
provider = method_options.get("provider", "openai")
env_key = f"{provider.upper()}_API_KEY"
provider_name = method_options.get("provider", "openai")
env_key = f"{provider_name.upper()}_API_KEY"
api_key = os.getenv(env_key)
if api_key:
method_options["api_key"] = api_key
@@ -440,6 +439,12 @@ class RelationExtractor:
if verbose_mode and method_name == "llm":
import sys
print(f" [RelationExtractor] Processing with {method_name}...", flush=True, file=sys.stdout)
print(f" [RelationExtractor Debug] method_options keys: {list(method_options.keys())}", flush=True, file=sys.stdout)
if "api_key" in method_options:
masked = method_options["api_key"][:4] + "..." if method_options["api_key"] else "None"
print(f" [RelationExtractor Debug] api_key present: {masked}", flush=True, file=sys.stdout)
else:
print(f" [RelationExtractor Debug] api_key NOT present", flush=True, file=sys.stdout)
relations = method_func(text, entities, **method_options)
@@ -482,6 +487,11 @@ class RelationExtractor:
except Exception as e:
self.logger.warning(f"Method {method_name} failed: {e}")
if verbose_mode:
import sys
print(f" [RelationExtractor] ERROR: Method {method_name} failed: {e}", flush=True, file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
continue
# Use first successful method or combine
@@ -239,6 +239,10 @@ class SemanticNetworkExtractor:
return idx, network
except Exception as e:
self.logger.warning(f"Failed to process item {idx}: {e}")
verbose_mode = kwargs.get("verbose", False) or self.config.get("verbose", False)
if verbose_mode:
import sys
print(f" [SemanticNetworkExtractor] ERROR: Batch item {idx} failed: {e}", flush=True, file=sys.stderr)
return idx, None
if max_workers > 1:
@@ -434,6 +438,12 @@ class SemanticNetworkExtractor:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
verbose_mode = options.get("verbose", False) or self.config.get("verbose", False)
if verbose_mode:
import sys
print(f" [SemanticNetworkExtractor] ERROR: Extraction failed: {e}", flush=True, file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
raise
def _build_network(
@@ -461,18 +461,28 @@ class TripletExtractor:
method_options["model"] = all_options.get(
"llm_model", all_options.get("model")
)
# Pass api_key if provided (needed for all providers)
if "api_key" in all_options:
method_options["api_key"] = all_options["api_key"]
elif "api_key" not in method_options:
# Try to get from environment as fallback
# Ensure api_key is populated: check explicitly provided or fallback to env
current_key = method_options.get("api_key")
if not current_key:
# Not found or empty/None, try environment
import os
provider = method_options.get("provider", "openai")
env_key = f"{provider.upper()}_API_KEY"
provider_name = method_options.get("provider", "openai")
env_key = f"{provider_name.upper()}_API_KEY"
api_key = os.getenv(env_key)
if api_key:
method_options["api_key"] = api_key
# Print progress if verbose mode is enabled (only for LLM method to avoid spam)
verbose_mode = options.get("verbose", False) or self.config.get("verbose", False)
if verbose_mode and method_name == "llm":
import sys
print(f" [TripletExtractor] Processing with {method_name}...", flush=True, file=sys.stdout)
if "api_key" in method_options:
masked = method_options["api_key"][:4] + "..." if method_options["api_key"] else "None"
print(f" [TripletExtractor Debug] api_key present: {masked}", flush=True, file=sys.stdout)
else:
print(f" [TripletExtractor Debug] api_key NOT present", flush=True, file=sys.stdout)
triplets = method_func(
text,
entities=entities,
@@ -480,6 +490,11 @@ class TripletExtractor:
**method_options,
)
# Print result count if verbose (only for LLM method)
if verbose_mode and method_name == "llm" and len(triplets) > 0:
import sys
print(f" [TripletExtractor] Extracted {len(triplets)} triplets", flush=True, file=sys.stdout)
# Apply weighted scoring if triplet_types are provided
if triplet_types:
try:
@@ -515,6 +530,12 @@ class TripletExtractor:
except Exception as e:
self.logger.warning(f"Method {method_name} failed: {e}")
verbose_mode = options.get("verbose", False) or self.config.get("verbose", False)
if verbose_mode:
import sys
print(f" [TripletExtractor] ERROR: Method {method_name} failed: {e}", flush=True, file=sys.stderr)
import traceback
traceback.print_exc(file=sys.stderr)
continue
# Use first successful method or fallback to relation conversion
+25 -4
View File
@@ -34,6 +34,7 @@ License: MIT
"""
import inspect
import os
import sys
import threading
import time
@@ -46,6 +47,13 @@ from typing import Any, Callable, Dict, List, Optional, Tuple, Union
from .logging import get_logger
DISABLE_JUPYTER_PROGRESS = os.getenv("SEMANTICA_DISABLE_JUPYTER_PROGRESS", "").strip().lower() in (
"1",
"true",
"yes",
"on",
)
# Try to import IPython for Jupyter support
try:
from IPython import get_ipython
@@ -1017,6 +1025,7 @@ class ProgressTracker:
# Detect environment - will be checked dynamically
self.is_jupyter = self._detect_jupyter()
self.disable_jupyter_progress = DISABLE_JUPYTER_PROGRESS
# Create displays
self.displays: List[ProgressDisplay] = []
@@ -1024,7 +1033,7 @@ class ProgressTracker:
# Always try Jupyter first if available, fallback to console
if IPYTHON_AVAILABLE:
# Try to detect Jupyter - if available, use it
if self.is_jupyter:
if self.is_jupyter and not self.disable_jupyter_progress:
self.displays.append(JupyterProgressDisplay(use_emoji=use_emoji))
# Also add console as fallback for immediate feedback
self.displays.append(
@@ -1213,7 +1222,11 @@ class ProgressTracker:
if IPYTHON_AVAILABLE and not self.is_jupyter:
self.is_jupyter = self._detect_jupyter()
# If Jupyter is now detected and we don't have a Jupyter display, add it
if self.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays):
if (
self.is_jupyter
and not self.disable_jupyter_progress
and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays)
):
# Insert Jupyter display at the beginning for priority
self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji))
@@ -1341,7 +1354,11 @@ class ProgressTracker:
if IPYTHON_AVAILABLE and not self.is_jupyter:
self.is_jupyter = self._detect_jupyter()
# If Jupyter is now detected and we don't have a Jupyter display, add it
if self.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays):
if (
self.is_jupyter
and not self.disable_jupyter_progress
and not any(isinstance(d, JupyterProgressDisplay) for d in self.displays)
):
# Insert Jupyter display at the beginning for priority
self.displays.insert(0, JupyterProgressDisplay(use_emoji=self.use_emoji))
@@ -1526,7 +1543,11 @@ def get_progress_tracker() -> ProgressTracker:
if IPYTHON_AVAILABLE and not _global_tracker.is_jupyter:
_global_tracker.is_jupyter = _global_tracker._detect_jupyter()
# If Jupyter is now detected and we don't have a Jupyter display, add it
if _global_tracker.is_jupyter and not any(isinstance(d, JupyterProgressDisplay) for d in _global_tracker.displays):
if (
_global_tracker.is_jupyter
and not _global_tracker.disable_jupyter_progress
and not any(isinstance(d, JupyterProgressDisplay) for d in _global_tracker.displays)
):
# Insert Jupyter display at the beginning for priority
_global_tracker.displays.insert(0, JupyterProgressDisplay(use_emoji=_global_tracker.use_emoji))
+138 -1
View File
@@ -38,6 +38,7 @@ License: MIT
"""
from typing import Any, Dict, List, Optional, Tuple, Union
import concurrent.futures
import numpy as np
@@ -61,7 +62,7 @@ class VectorStore:
SUPPORTED_BACKENDS = {"faiss", "weaviate", "qdrant", "milvus", "inmemory"}
def __init__(self, backend="faiss", config=None, **kwargs):
def __init__(self, backend="faiss", config=None, max_workers: int = 6, **kwargs):
"""Initialize vector store."""
if backend.lower() not in self.SUPPORTED_BACKENDS:
raise ValueError(
@@ -72,6 +73,7 @@ class VectorStore:
self.logger = get_logger("vector_store")
self.config = config or {}
self.config.update(kwargs)
self.max_workers = max_workers
self.progress_tracker = get_progress_tracker()
# Ensure progress tracker is enabled
if not self.progress_tracker.enabled:
@@ -127,6 +129,141 @@ class VectorStore:
self.logger.warning("Using random fallback embedding")
return np.random.rand(self.dimension).astype(np.float32)
def embed_batch(self, texts: List[str]) -> List[np.ndarray]:
"""
Generate embeddings for a list of texts using the internal embedder.
Args:
texts: List of texts to embed
Returns:
List of numpy arrays
"""
if self.embedder:
try:
# generate_embeddings handles list input
embeddings = self.embedder.generate_embeddings(texts)
# Ensure it returns a list of arrays (it returns 2D array or list)
if isinstance(embeddings, np.ndarray):
return list(embeddings)
return embeddings
except Exception as e:
self.logger.warning(f"Batch embedding generation failed: {e}")
# Fallback
self.logger.warning("Using random fallback embeddings for batch")
return [np.random.rand(self.dimension).astype(np.float32) for _ in texts]
def add_documents(
self,
documents: List[str],
metadata: Optional[List[Dict[str, Any]]] = None,
batch_size: int = 32,
parallel: bool = True,
**options,
) -> List[str]:
"""
Add multiple documents to the store with parallel embedding generation.
Args:
documents: List of document texts
metadata: List of metadata dictionaries
batch_size: Number of documents to process in one batch
parallel: Whether to use parallel processing for embeddings
**options: Additional options
Returns:
List[str]: Vector IDs
"""
if not documents:
return []
num_docs = len(documents)
metadata = metadata or [{} for _ in range(num_docs)]
if len(metadata) != num_docs:
raise ValueError("Metadata list length must match documents length")
all_vectors = [None] * num_docs
# Helper for processing a batch
def process_batch(start_idx: int, end_idx: int):
batch_texts = documents[start_idx:end_idx]
batch_embeddings = self.embed_batch(batch_texts)
return start_idx, batch_embeddings
# Calculate batches
batches = []
for i in range(0, num_docs, batch_size):
batches.append((i, min(i + batch_size, num_docs)))
tracking_id = self.progress_tracker.start_tracking(
module="vector_store",
submodule="VectorStore",
message=f"Processing {num_docs} documents (parallel={parallel})",
)
try:
if parallel and self.max_workers > 1:
self.progress_tracker.update_tracking(
tracking_id, message=f"Embedding with {self.max_workers} workers..."
)
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
futures = [
executor.submit(process_batch, start, end)
for start, end in batches
]
completed = 0
for future in concurrent.futures.as_completed(futures):
start_idx, embeddings = future.result()
# Place results in correct order
for i, emb in enumerate(embeddings):
all_vectors[start_idx + i] = emb
completed += 1
if completed % 5 == 0: # Update progress periodically
self.progress_tracker.update_tracking(
tracking_id,
message=f"Embedded batch {completed}/{len(batches)}"
)
else:
# Sequential processing
self.progress_tracker.update_tracking(
tracking_id, message="Embedding sequentially..."
)
for i, (start, end) in enumerate(batches):
_, embeddings = process_batch(start, end)
for j, emb in enumerate(embeddings):
all_vectors[start + j] = emb
if i % 5 == 0:
self.progress_tracker.update_tracking(
tracking_id,
message=f"Embedded batch {i+1}/{len(batches)}"
)
# Verify all embeddings generated
if any(v is None for v in all_vectors):
raise ProcessingError("Failed to generate all embeddings")
# Store all vectors in one go
self.progress_tracker.update_tracking(tracking_id, message="Storing vectors...")
vector_ids = self.store_vectors(all_vectors, metadata=metadata, **options)
self.progress_tracker.stop_tracking(
tracking_id,
status="completed",
message=f"Added {len(vector_ids)} documents",
)
return vector_ids
except Exception as e:
self.progress_tracker.stop_tracking(
tracking_id, status="failed", message=str(e)
)
raise
def store(
self,
vectors: List[np.ndarray],
+214
View File
@@ -0,0 +1,214 @@
import os
import shutil
import tempfile
import pytest
from pathlib import Path
from semantica.ingest import OntologyIngestor, ingest, ingest_ontology, OntologyData
class TestOntologyIngestor:
@pytest.fixture
def sample_ttl_content(self):
return """
@prefix : <http://example.org/ontology/> .
@prefix owl: <http://www.w3.org/2002/07/owl#> .
@prefix rdf: <http://www.w3.org/1999/02/22-rdf-syntax-ns#> .
@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .
@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .
<http://example.org/ontology/> rdf:type owl:Ontology ;
rdfs:label "Test Ontology" .
:Person rdf:type owl:Class ;
rdfs:label "Person" .
:hasName rdf:type owl:DatatypeProperty ;
rdfs:domain :Person ;
rdfs:range xsd:string .
"""
def test_ingest_single_file(self, sample_ttl_content):
with tempfile.NamedTemporaryFile(delete=False, suffix=".ttl", mode="w") as tmp:
tmp.write(sample_ttl_content)
tmp_path = tmp.name
try:
ingestor = OntologyIngestor()
result = ingestor.ingest_ontology(tmp_path)
assert isinstance(result, OntologyData)
assert result.data["name"] == "Test Ontology" or result.data["name"] == os.path.basename(tmp_path)
assert any(cls["name"] == "Person" for cls in result.data["classes"])
assert any(prop["name"] == "hasName" for prop in result.data["properties"])
assert result.metadata["format"] == "ttl" or result.metadata["format"] == "turtle"
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_ingest_directory(self, sample_ttl_content):
with tempfile.TemporaryDirectory() as tmp_dir:
# Create two ontology files
file1 = os.path.join(tmp_dir, "ont1.ttl")
file2 = os.path.join(tmp_dir, "ont2.rdf")
with open(file1, "w") as f:
f.write(sample_ttl_content)
# Simple RDF/XML content for the second file
rdf_content = """
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
xmlns:owl="http://www.w3.org/2002/07/owl#">
<owl:Ontology rdf:about="http://example.org/ont2"/>
<owl:Class rdf:about="http://example.org/ont2/Animal"/>
</rdf:RDF>
"""
with open(file2, "w") as f:
f.write(rdf_content)
ingestor = OntologyIngestor()
results = ingestor.ingest_directory(tmp_dir)
assert len(results) == 2
assert all(isinstance(r, OntologyData) for r in results)
# Verify results contain expected classes
classes = [cls["name"] for res in results for cls in res.data["classes"]]
assert "Person" in classes
assert "Animal" in classes
def test_unified_ingest_function(self, sample_ttl_content):
with tempfile.NamedTemporaryFile(delete=False, suffix=".ttl", mode="w") as tmp:
tmp.write(sample_ttl_content)
tmp_path = tmp.name
try:
# Test auto-detection via unified ingest
result = ingest(tmp_path)
assert "ontology" in result
assert isinstance(result["ontology"], OntologyData)
assert len(result["ontology"].data["classes"]) > 0
# Test explicit source type
result_explicit = ingest(tmp_path, source_type="ontology")
assert "ontology" in result_explicit
assert result_explicit["ontology"].metadata["source_path"] == tmp_path
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_convenience_function(self, sample_ttl_content):
with tempfile.NamedTemporaryFile(delete=False, suffix=".n3", mode="w") as tmp:
tmp.write(sample_ttl_content)
tmp_path = tmp.name
try:
result = ingest_ontology(tmp_path)
assert isinstance(result, OntologyData)
assert len(result.data["classes"]) > 0
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_ingest_formats(self):
"""Test ingestion of all supported formats."""
ingestor = OntologyIngestor()
# 1. JSON-LD
jsonld_content = """
{
"@context": {
"owl": "http://www.w3.org/2002/07/owl#",
"rdf": "http://www.w3.org/1999/02/22-rdf-syntax-ns#",
"rdfs": "http://www.w3.org/2000/01/rdf-schema#"
},
"@id": "http://example.org/jsonld",
"@type": "owl:Ontology",
"rdfs:label": "JSON-LD Ontology",
"owl:versionInfo": "1.0"
}
"""
with tempfile.NamedTemporaryFile(delete=False, suffix=".jsonld", mode="w") as tmp:
tmp.write(jsonld_content)
tmp_path = tmp.name
try:
result = ingestor.ingest_ontology(tmp_path)
assert result.data["name"] == "JSON-LD Ontology"
assert result.metadata["format"] == "json-ld"
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
# 2. N-Triples
nt_content = '<http://example.org/nt/Class> <http://www.w3.org/1999/02/22-rdf-syntax-ns#type> <http://www.w3.org/2002/07/owl#Class> .\n'
with tempfile.NamedTemporaryFile(delete=False, suffix=".nt", mode="w") as tmp:
tmp.write(nt_content)
tmp_path = tmp.name
try:
result = ingestor.ingest_ontology(tmp_path)
# N-Triples often doesn't have ontology metadata, so name might default to basename
assert result.data["name"] == os.path.basename(tmp_path)
assert any(cls["uri"] == "http://example.org/nt/Class" for cls in result.data["classes"])
assert result.metadata["format"] == "nt"
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
# 3. Notation3
n3_content = """
@prefix : <http://example.org/n3/> .
@prefix owl: <http://www.w3.org/2002/07/owl#> .
:N3Class a owl:Class .
"""
with tempfile.NamedTemporaryFile(delete=False, suffix=".n3", mode="w") as tmp:
tmp.write(n3_content)
tmp_path = tmp.name
try:
result = ingestor.ingest_ontology(tmp_path)
assert any(cls["uri"] == "http://example.org/n3/N3Class" for cls in result.data["classes"])
# format might be 'n3' or 'turtle' depending on rdflib detection as they are similar
assert result.metadata["format"] in ["n3", "turtle"]
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
# 4. RDF/XML (.owl)
owl_content = """
<rdf:RDF xmlns:rdf="http://www.w3.org/1999/02/22-rdf-syntax-ns#"
xmlns:owl="http://www.w3.org/2002/07/owl#"
xmlns:rdfs="http://www.w3.org/2000/01/rdf-schema#">
<owl:Ontology rdf:about="http://example.org/owl"/>
<owl:Class rdf:about="http://example.org/owl/OwlClass">
<rdfs:label>OwlClass</rdfs:label>
</owl:Class>
</rdf:RDF>
"""
with tempfile.NamedTemporaryFile(delete=False, suffix=".owl", mode="w") as tmp:
tmp.write(owl_content)
tmp_path = tmp.name
try:
result = ingestor.ingest_ontology(tmp_path)
assert any(cls["name"] == "OwlClass" for cls in result.data["classes"])
assert result.metadata["format"] in ["xml", "rdf", "owl"]
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
def test_error_handling(self):
ingestor = OntologyIngestor()
with pytest.raises(Exception): # Specific exception type depends on implementation, likely ValidationError or FileNotFoundError
ingestor.ingest_ontology("non_existent_file.ttl")
def test_invalid_content(self):
with tempfile.NamedTemporaryFile(delete=False, suffix=".ttl", mode="w") as tmp:
tmp.write("This is not valid turtle content")
tmp_path = tmp.name
try:
ingestor = OntologyIngestor()
# Depending on implementation, this might raise an exception or return partial/empty result with error in metadata
# Given current implementation uses g.parse(), it likely raises an exception which is caught or propagated
# If propagated:
with pytest.raises(Exception):
ingestor.ingest_ontology(tmp_path)
finally:
if os.path.exists(tmp_path):
os.remove(tmp_path)
+98
View File
@@ -0,0 +1,98 @@
import os
import unittest
import time
from semantica.semantic_extract import NERExtractor, RelationExtractor
# Use environment variable for API key
_GROQ_KEY = os.getenv("GROQ_API_KEY") or os.getenv("GROQ_TEST_API_KEY")
@unittest.skipUnless(_GROQ_KEY, "Groq key not set; skipping live integration test")
class TestGroqRelationsIntegration(unittest.TestCase):
def setUp(self):
self.api_key = _GROQ_KEY
self.model = "llama-3.1-8b-instant"
# Short, unambiguous finance snippet
self.text_short = (
"Apple reported revenue of $4.4 billion in Q1 2024 and provided guidance for FY 2025."
)
# Longer text to exercise chunking and ensure no hang
self.text_long = (
"Apple reported revenue of $4.4 billion in Q1 2024. "
"The company also reported growth of 12% year-over-year and provided guidance for FY 2025. "
"Microsoft reported revenue of $6.1 billion in Q2 2024 and expects sequential growth. "
"NVIDIA reported record revenue in 2024 Q1 and guided for higher revenue in Q2 2024. "
) * 20 # expand length
def _extract_entities(self, text):
ner = NERExtractor(
method="llm",
provider="groq",
llm_model=self.model,
api_key=self.api_key,
temperature=0.0,
)
entities = ner.extract_entities(text, entity_types=["ORGANIZATION", "MONEY", "DATE", "EVENT", "PERCENT"])
self.assertIsInstance(entities, list)
return entities
def test_relations_short_text(self):
entities = self._extract_entities(self.text_short)
self.assertGreater(len(entities), 0, "NER should extract entities for short text")
relation_extractor = RelationExtractor(
method="llm",
relation_types=[
"HAS_REVENUE",
"HAS_GROWTH",
"PROVIDES_GUIDANCE",
"IN_QUARTER",
"FOR_PERIOD",
"RELATED_TO",
],
provider="groq",
llm_model=self.model,
api_key=self.api_key,
temperature=0.0,
verbose=True,
)
start = time.time()
relations = relation_extractor.extract_relations(text=self.text_short, entities=entities)
elapsed = time.time() - start
self.assertIsInstance(relations, list)
# Ensure call completes reasonably fast (network dependent; allow generous bound)
self.assertLess(elapsed, 60, f"Extraction took too long: {elapsed:.2f}s")
# Do not strictly assert >0 as model output may vary, but log for diagnostics
if relations:
sample = relations[0]
self.assertTrue(hasattr(sample, "subject") and hasattr(sample, "predicate") and hasattr(sample, "object"))
def test_relations_long_text_chunking(self):
entities = self._extract_entities(self.text_long)
self.assertGreater(len(entities), 0, "NER should extract entities for long text")
relation_extractor = RelationExtractor(
method="llm",
relation_types=["RELATED_TO", "HAS_REVENUE", "IN_QUARTER"],
provider="groq",
llm_model=self.model,
api_key=self.api_key,
temperature=0.0,
verbose=True,
)
start = time.time()
relations = relation_extractor.extract_relations(text=self.text_long, entities=entities)
elapsed = time.time() - start
self.assertIsInstance(relations, list)
# Ensure completion (chunked path) and no hang
self.assertLess(elapsed, 120, f"Chunked extraction took too long: {elapsed:.2f}s")
if relations:
for r in relations[:3]:
self.assertTrue(hasattr(r, "subject") and hasattr(r, "predicate") and hasattr(r, "object"))
if __name__ == "__main__":
unittest.main()
+232
View File
@@ -0,0 +1,232 @@
import unittest
from unittest.mock import MagicMock, patch
from semantica.kg.graph_builder import GraphBuilder
class TestGraphBuilderExternal(unittest.TestCase):
def setUp(self):
self.mock_tracker_patcher = patch("semantica.utils.progress_tracker.get_progress_tracker")
self.mock_get_tracker = self.mock_tracker_patcher.start()
self.mock_tracker = MagicMock()
self.mock_get_tracker.return_value = self.mock_tracker
self.mock_resolver_patcher = patch("semantica.kg.entity_resolver.EntityResolver")
self.mock_resolver_cls = self.mock_resolver_patcher.start()
self.mock_conflict_patcher = patch("semantica.conflicts.conflict_detector.ConflictDetector")
self.mock_conflict_cls = self.mock_conflict_patcher.start()
def tearDown(self):
self.mock_tracker_patcher.stop()
self.mock_resolver_patcher.stop()
self.mock_conflict_patcher.stop()
def test_single_source_dict_with_source_id_target_id(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
entities = [
{"id": "drug:1", "name": "Aspirin", "type": "Drug"},
{"id": "disease:1", "name": "Myocardial infarction", "type": "Disease"},
]
relationships = [
{"source_id": "drug:1", "target_id": "disease:1", "type": "TREATS"},
]
source = {"entities": entities, "relationships": relationships}
kg = builder.build(source)
self.assertEqual(len(kg["entities"]), 2)
self.assertEqual(len(kg["relationships"]), 1)
rel = kg["relationships"][0]
self.assertEqual(rel.get("source"), "drug:1")
self.assertEqual(rel.get("target"), "disease:1")
self.assertEqual(kg["metadata"]["num_relationships"], 1)
def test_sources_list_merge_with_external_relationships(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source1 = {
"entities": [{"id": "1", "name": "A"}],
"relationships": [{"source_id": "1", "target_id": "2", "type": "REL_1"}],
}
source2 = {
"entities": [{"id": "2", "name": "B"}],
"relationships": [{"source_id": "2", "target_id": "1", "type": "REL_2"}],
}
kg = builder.build([source1, source2])
self.assertEqual(len(kg["entities"]), 2)
self.assertEqual(len(kg["relationships"]), 2)
sources = {r["source"] for r in kg["relationships"]}
targets = {r["target"] for r in kg["relationships"]}
self.assertEqual(sources, {"1", "2"})
self.assertEqual(targets, {"1", "2"})
def test_build_with_explicit_relationships_argument_external_ids(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
entities = [
{"id": "1", "name": "A"},
{"id": "2", "name": "B"},
]
relationships = [
{"source_id": "1", "target_id": "2", "type": "REL"},
]
kg = builder.build(entities, relationships=relationships)
self.assertEqual(len(kg["entities"]), 2)
self.assertEqual(len(kg["relationships"]), 1)
rel = kg["relationships"][0]
self.assertEqual(rel.get("source"), "1")
self.assertEqual(rel.get("target"), "2")
def test_build_single_source_external_graph(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source = {
"entities": [{"id": "1", "name": "A"}],
"relationships": [{"source_id": "1", "target_id": "1", "type": "SELF"}],
}
kg = builder.build_single_source(source)
self.assertEqual(len(kg["entities"]), 1)
self.assertEqual(len(kg["relationships"]), 1)
rel = kg["relationships"][0]
self.assertEqual(rel.get("source"), "1")
self.assertEqual(rel.get("target"), "1")
def test_relationship_key_variants_normalized(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
entities = [
{"id": "1", "name": "A"},
{"id": "2", "name": "B"},
{"id": "3", "name": "C"},
{"id": "4", "name": "D"},
]
relationships = [
{"source_id": "1", "target_id": "2", "type": "R1"},
{"source": "2", "target": "3", "type": "R2"},
{"subject": "3", "object": "4", "type": "R3"},
]
kg = builder.build({"entities": entities, "relationships": relationships})
self.assertEqual(len(kg["relationships"]), 3)
ids = {(r["source"], r["target"]) for r in kg["relationships"]}
self.assertIn(("1", "2"), ids)
self.assertIn(("2", "3"), ids)
self.assertIn(("3", "4"), ids)
def test_warning_when_all_relationships_dropped(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source = {
"entities": [],
"relationships": [{"foo": "x"}, {"bar": "y"}],
}
with patch.object(builder.logger, "warning") as mock_warning:
kg = builder.build(source)
self.assertEqual(len(kg["relationships"]), 0)
mock_warning.assert_called()
args, _ = mock_warning.call_args
self.assertIn("All relationships were dropped", args[0])
def test_no_warning_when_some_relationships_kept(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source = {
"entities": [{"id": "1"}, {"id": "2"}],
"relationships": [
{"source_id": "1", "target_id": "2", "type": "REL"},
{"foo": "x"},
],
}
with patch.object(builder.logger, "warning") as mock_warning:
kg = builder.build(source)
self.assertEqual(len(kg["relationships"]), 2)
mock_warning.assert_not_called()
def test_issue_208_minimal_reproduction_shape(self):
builder = GraphBuilder(
merge_entities=False,
entity_resolution_strategy="none",
resolve_conflicts=False,
)
entities = [
{"id": "e1", "name": "Entity 1"},
{"id": "e2", "name": "Entity 2"},
{"id": "e3", "name": "Entity 3"},
]
relationships = [
{"source_id": "e1", "target_id": "e2", "type": "REL_1"},
{"source_id": "e2", "target_id": "e3", "type": "REL_2"},
]
entity_ids = {e["id"] for e in entities}
for r in relationships:
self.assertIn(r["source_id"], entity_ids)
self.assertIn(r["target_id"], entity_ids)
kg = builder.build(
sources=[{"entities": entities, "relationships": relationships}],
merge_entities=False,
)
self.assertEqual(len(kg["entities"]), 3)
self.assertEqual(len(kg["relationships"]), 2)
pairs = {(r["source"], r["target"]) for r in kg["relationships"]}
self.assertIn(("e1", "e2"), pairs)
self.assertIn(("e2", "e3"), pairs)
def test_issue_206_earnings_call_shape(self):
builder = GraphBuilder(
merge_entities=False,
entity_resolution_strategy="none",
resolve_conflicts=False,
)
entities = [
{
"id": "entity_446_MDA Space Ltd.",
"name": "MDA Space Ltd.",
"type": "ORGANIZATION",
},
{
"id": "entity_500_$409.8 million",
"name": "$409.8 million",
"type": "MONEY",
},
]
relationships = [
{
"id": None,
"source_id": "MDA Space Ltd.",
"target_id": "$409.8 million",
"type": "HAS_REVENUE",
"confidence": 0.975,
"metadata": {},
}
]
kg = builder.build(
sources=[{"entities": entities, "relationships": relationships}],
merge_entities=False,
)
self.assertEqual(len(kg["entities"]), 2)
self.assertEqual(len(kg["relationships"]), 1)
rel = kg["relationships"][0]
self.assertEqual(rel.get("source"), "MDA Space Ltd.")
self.assertEqual(rel.get("target"), "$409.8 million")
+68
View File
@@ -91,6 +91,30 @@ class TestGraphBuilder(unittest.TestCase):
graph2 = builder.build(source_list)
self.assertEqual(len(graph2["entities"]), 2)
def test_build_with_external_relationship_ids(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
entities = [
{"id": "1", "name": "A"},
{"id": "2", "name": "B"},
]
relationships = [
{"source_id": "1", "target_id": "2", "type": "rel"},
]
source = {
"entities": entities,
"relationships": relationships,
}
graph = builder.build(source)
self.assertEqual(len(graph["entities"]), 2)
self.assertEqual(len(graph["relationships"]), 1)
rel = graph["relationships"][0]
self.assertEqual(rel.get("source"), "1")
self.assertEqual(rel.get("target"), "2")
def test_build_with_conflict_resolution(self):
"""Test building with conflict resolution enabled"""
builder = GraphBuilder(resolve_conflicts=True)
@@ -106,6 +130,50 @@ class TestGraphBuilder(unittest.TestCase):
self.mock_conflict_cls.return_value.detect_conflicts.assert_called_once()
self.mock_conflict_cls.return_value.resolve_conflicts.assert_called_once()
def test_build_single_source(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source = {
"entities": [{"id": "1", "name": "A"}],
"relationships": [{"source_id": "1", "target_id": "1", "type": "self"}],
}
graph = builder.build_single_source(source)
self.assertEqual(len(graph["entities"]), 1)
self.assertEqual(len(graph["relationships"]), 1)
def test_build_with_explicit_relationships_argument(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
entities = [
{"id": "1", "name": "A"},
{"id": "2", "name": "B"},
]
relationships = [
{"source_id": "1", "target_id": "2", "type": "rel"},
]
graph = builder.build(entities, relationships=relationships)
self.assertEqual(len(graph["entities"]), 2)
self.assertEqual(len(graph["relationships"]), 1)
rel = graph["relationships"][0]
self.assertEqual(rel.get("source"), "1")
self.assertEqual(rel.get("target"), "2")
def test_build_warns_when_all_relationships_dropped(self):
builder = GraphBuilder(merge_entities=False, resolve_conflicts=False)
source = {
"entities": [],
"relationships": [{"foo": "x"}, {"bar": "y"}],
}
with patch.object(builder.logger, "warning") as mock_warning:
graph = builder.build(source)
self.assertEqual(len(graph["relationships"]), 0)
mock_warning.assert_called()
args, _ = mock_warning.call_args
self.assertIn("All relationships were dropped", args[0])
class TestGraphAnalyzer(unittest.TestCase):
def setUp(self):
self.mock_tracker_patcher = patch("semantica.kg.graph_analyzer.get_progress_tracker")
+20
View File
@@ -107,5 +107,25 @@ class TestPipelineModule(unittest.TestCase):
self.assertEqual(execution_order, ["A", "B", "C"])
def test_imports_no_circular_dependencies(self):
import semantica
_ = semantica.pipeline
from semantica.pipeline import PipelineBuilder, PipelineValidator
from semantica.deduplication import DuplicateDetector
builder = PipelineBuilder()
builder.add_step("step1", "dummy")
pipeline = builder.build("import_test_pipeline")
validator = PipelineValidator()
result = validator.validate_pipeline(pipeline)
self.assertTrue(result.valid)
detector = DuplicateDetector()
self.assertIsNotNone(detector)
if __name__ == "__main__":
unittest.main()
+118
View File
@@ -0,0 +1,118 @@
import unittest
from semantica.semantic_extract.methods import extract_relations_llm
from semantica.semantic_extract.ner_extractor import Entity
class FakeProvider:
def __init__(self, typed_payload=None, structured_payload=None):
self._typed_payload = typed_payload
self._structured_payload = structured_payload
def is_available(self):
return True
# Simulate typed output return: can be dict or an object with relations
def generate_typed(self, prompt, schema, **kwargs):
return self._typed_payload if self._typed_payload is not None else {"relations": []}
def generate_structured(self, prompt, **kwargs):
return self._structured_payload if self._structured_payload is not None else {"relations": []}
class TestLLMRelationExtraction(unittest.TestCase):
def setUp(self):
# Minimal realistic text and entities
self.text = "Apple reported revenue of $4.4 billion in Q1 2024."
self.entities = [
Entity(text="Apple", label="ORGANIZATION", start_char=0, end_char=5, confidence=0.99),
Entity(text="$4.4 billion", label="MONEY", start_char=26, end_char=39, confidence=0.99),
Entity(text="Q1 2024", label="DATE", start_char=43, end_char=51, confidence=0.99),
]
def _monkeypatch_provider(self, provider_instance):
# Monkeypatch create_provider used by extract_relations_llm
import semantica.semantic_extract.methods as methods
self._orig_create_provider = methods.create_provider
def _fake_create_provider(provider, model=None, **kwargs):
return provider_instance
methods.create_provider = _fake_create_provider
def tearDown(self):
# Restore original create_provider if patched
try:
import semantica.semantic_extract.methods as methods
if hasattr(self, "_orig_create_provider"):
methods.create_provider = self._orig_create_provider
except Exception:
pass
def test_typed_relations_parsed(self):
# Typed returns a dict compatible with parser
typed_payload = {
"relations": [
{
"subject": "Apple",
"predicate": "HAS_REVENUE",
"object": "$4.4 billion",
"confidence": 0.92,
},
{
"subject": "Apple",
"predicate": "IN_QUARTER",
"object": "Q1 2024",
"confidence": 0.9,
},
]
}
fake = FakeProvider(typed_payload=typed_payload)
self._monkeypatch_provider(fake)
rels = extract_relations_llm(
text=self.text,
entities=self.entities,
provider="groq",
model="llama-3.1-8b-instant",
relation_types=["HAS_REVENUE", "IN_QUARTER"],
verbose=True,
)
self.assertGreaterEqual(len(rels), 2, "Expected at least two relations from typed payload")
preds = {(r.subject.text, r.predicate, r.object.text) for r in rels}
self.assertIn(("Apple", "HAS_REVENUE", "$4.4 billion"), preds)
self.assertIn(("Apple", "IN_QUARTER", "Q1 2024"), preds)
def test_structured_fallback_used(self):
# Typed returns zero, structured has content
typed_payload = {"relations": []}
structured_payload = {
"relations": [
{
"subject": "Apple",
"predicate": "HAS_REVENUE",
"object": "$4.4 billion",
"confidence": 0.88,
}
]
}
fake = FakeProvider(typed_payload=typed_payload, structured_payload=structured_payload)
self._monkeypatch_provider(fake)
rels = extract_relations_llm(
text=self.text,
entities=self.entities,
provider="groq",
model="llama-3.1-8b-instant",
relation_types=["HAS_REVENUE"],
verbose=True,
)
self.assertEqual(len(rels), 1, "Expected fallback to structured JSON to yield one relation")
r = rels[0]
self.assertEqual(r.subject.text, "Apple")
self.assertEqual(r.object.text, "$4.4 billion")
self.assertEqual(r.predicate, "HAS_REVENUE")
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,149 @@
import unittest
import numpy as np
import time
import sys
import os
import logging
from unittest.mock import MagicMock, patch
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../..")))
from semantica.vector_store import VectorStore
from semantica.utils.exceptions import ProcessingError
class TestVectorStoreParallel(unittest.TestCase):
def setUp(self):
logging.getLogger("vector_store").setLevel(logging.ERROR)
self.dimension = 4
self.store = VectorStore(
backend="inmemory",
dimension=self.dimension,
)
self.store.embedder = MagicMock()
def test_embed_batch_success(self):
texts = ["a", "b", "c"]
expected_embeddings = [
np.array([0.1] * 4, dtype=np.float32),
np.array([0.2] * 4, dtype=np.float32),
np.array([0.3] * 4, dtype=np.float32),
]
self.store.embedder.generate_embeddings.return_value = expected_embeddings
results = self.store.embed_batch(texts)
self.assertEqual(len(results), 3)
self.assertTrue(np.allclose(results[0], expected_embeddings[0]))
self.store.embedder.generate_embeddings.assert_called_once_with(texts)
def test_embed_batch_fallback(self):
texts = ["a", "b"]
self.store.embedder.generate_embeddings.side_effect = Exception("Model error")
results = self.store.embed_batch(texts)
self.assertEqual(len(results), 2)
self.assertEqual(results[0].shape, (self.dimension,))
self.assertTrue(isinstance(results[0], np.ndarray))
def test_add_documents_empty(self):
ids = self.store.add_documents([])
self.assertEqual(ids, [])
def test_add_documents_metadata_mismatch(self):
with self.assertRaises(ValueError):
self.store.add_documents(["doc1"], metadata=[{}, {}])
def test_add_documents_parallel_success(self):
num_docs = 10
documents = [f"doc_{i}" for i in range(num_docs)]
metadata = [{"id": i} for i in range(num_docs)]
def mock_embed_batch(texts):
return [np.full(self.dimension, float(i)) for i, _ in enumerate(texts)]
with patch.object(self.store, "embed_batch", side_effect=mock_embed_batch):
ids = self.store.add_documents(
documents,
metadata,
batch_size=2,
parallel=True,
)
self.assertEqual(len(ids), num_docs)
self.assertEqual(len(self.store.vectors), num_docs)
for i, vec_id in enumerate(ids):
stored_meta = self.store.get_metadata(vec_id)
self.assertEqual(stored_meta["id"], i)
def test_add_documents_sequential_success(self):
num_docs = 5
documents = [f"doc_{i}" for i in range(num_docs)]
with patch.object(self.store, "embed_batch") as mock_batch:
mock_batch.return_value = [np.zeros(self.dimension) for _ in range(num_docs)]
ids = self.store.add_documents(documents, parallel=False)
self.assertEqual(len(ids), num_docs)
self.assertEqual(mock_batch.call_count, 1)
def test_add_documents_error_propagation(self):
documents = ["doc1", "doc2"]
with patch.object(self.store, "embed_batch", side_effect=ValueError("Embedding Error")):
with self.assertRaises(Exception):
self.store.add_documents(documents, parallel=True)
def test_performance_simulation(self):
num_batches = 4
batch_delay = 0.1
batch_size = 1
documents = [f"doc_{i}" for i in range(num_batches)]
def slow_embed(texts):
time.sleep(batch_delay)
return [np.zeros(self.dimension) for _ in texts]
with patch.object(self.store, "embed_batch", side_effect=slow_embed):
start_seq = time.time()
self.store.add_documents(documents, batch_size=batch_size, parallel=False)
dur_seq = time.time() - start_seq
self.store.vectors = {}
start_par = time.time()
self.store.add_documents(documents, batch_size=batch_size, parallel=True)
dur_par = time.time() - start_par
print(f"\nPerformance Test:")
print(f"Sequential Duration: {dur_seq:.4f}s")
print(f"Parallel Duration: {dur_par:.4f}s")
print(f"Speedup: {dur_seq / dur_par:.2f}x")
self.assertLess(dur_par, dur_seq * 0.7)
def test_add_documents_batch_size_edge_cases(self):
documents = ["a", "b", "c"]
with patch.object(self.store, "embed_batch") as mock_batch:
mock_batch.side_effect = lambda texts: [np.zeros(4) for _ in texts]
self.store.add_documents(documents, batch_size=100)
self.assertEqual(mock_batch.call_count, 1)
mock_batch.reset_mock()
self.store.add_documents(documents, batch_size=1)
self.assertEqual(mock_batch.call_count, 3)
if __name__ == "__main__":
unittest.main()