Add Novita AI provider integration

- Add NovitaProvider class implementing OpenAI-compatible API
- Support for Novita AI API endpoint (https://api.novita.ai/openai)
- Configure via NOVITA_API_KEY environment variable or constructor
- Register 'novita' as built-in provider
- Update config.py to load NOVITA_API_KEY from environment
- Add test_novita_integration.py for provider testing

Default model: deepseek/deepseek-v3.2
This commit is contained in:
Alex-wuhu
2026-03-15 00:33:28 +08:00
parent c077944457
commit de03d05600
3 changed files with 146 additions and 2 deletions
+2 -2
View File
@@ -6,7 +6,7 @@ supporting multiple configuration sources including environment variables, confi
and programmatic configuration.
Supported Configuration Sources:
- Environment variables: OPENAI_API_KEY, GEMINI_API_KEY, GROQ_API_KEY, etc.
- Environment variables: OPENAI_API_KEY, GEMINI_API_KEY, GROQ_API_KEY, NOVITA_API_KEY, etc.
- Config files: YAML, JSON, TOML formats
- Programmatic: Python API for setting provider configurations
@@ -97,7 +97,7 @@ class Config:
def _load_env_vars(self):
"""Load configuration from environment variables."""
# Common environment variable patterns
providers = ["openai", "gemini", "groq", "anthropic", "ollama"]
providers = ["openai", "gemini", "groq", "anthropic", "ollama", "novita"]
for provider in providers:
env_key = f"{provider.upper()}_API_KEY"
api_key = os.getenv(env_key)
+60
View File
@@ -988,6 +988,65 @@ class DeepSeekProvider(BaseProvider):
except Exception as e:
raise ProcessingError(f"Failed to parse JSON from DeepSeek response: {e}")
class NovitaProvider(BaseProvider):
"""Novita AI provider implementation - OpenAI-compatible API."""
def __init__(self, api_key: Optional[str] = None, model: str = "deepseek/deepseek-v3.2", **kwargs):
"""Initialize Novita provider."""
super().__init__(**kwargs)
self.api_key = api_key or config.get_api_key("novita")
self.model = model
self.base_url = "https://api.novita.ai/openai"
self.client = None
self._init_client()
def _init_client(self):
try:
from openai import OpenAI
if self.api_key:
self.client = OpenAI(api_key=self.api_key, base_url=self.base_url)
except (ImportError, OSError):
self.client = None
self.logger.warning(
"openai library not installed. Install with: pip install semantica[llm-openai]"
)
def is_available(self) -> bool:
return self.client is not None
def generate(self, prompt: str, **kwargs) -> str:
if not self.client:
raise ProcessingError("Novita client not initialized. Set NOVITA_API_KEY or pass api_key.")
create_kwargs = {
"model": kwargs.get("model", self.model),
"messages": [{"role": "user", "content": prompt}],
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
response = self.client.chat.completions.create(**create_kwargs)
return response.choices[0].message.content
def generate_structured(self, prompt: str, **kwargs) -> Union[dict, list]:
"""Generate structured output."""
if not self.client:
raise ProcessingError("Novita client not initialized.")
create_kwargs = {
"model": kwargs.get("model", self.model),
"messages": [{"role": "user", "content": prompt}],
"response_format": {"type": "json_object"},
}
self._add_if_set(create_kwargs, kwargs, "temperature", "max_tokens")
response = self.client.chat.completions.create(**create_kwargs)
try:
return self._parse_json(response.choices[0].message.content)
except Exception as e:
raise ProcessingError(f"Failed to parse JSON from Novita response: {e}")
class HuggingFaceLLMProvider(BaseProvider):
"""HuggingFace transformers for LLM tasks."""
@@ -1370,6 +1429,7 @@ class ProviderPool:
"ollama": OllamaProvider,
"huggingface_llm": HuggingFaceLLMProvider,
"deepseek": DeepSeekProvider,
"novita": NovitaProvider,
}
provider_class = builtin.get(name.lower())
+84
View File
@@ -0,0 +1,84 @@
import os
import sys
import json
from pprint import pprint
# Ensure the package is in the path
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from semantica.semantic_extract.methods import (
extract_entities_llm,
extract_relations_llm,
extract_triplets_llm
)
from semantica.semantic_extract.providers import create_provider
from semantica.utils.exceptions import ProcessingError
# Set the API key
# Set the API key from environment
# We recommend setting it as an environment variable NOVITA_API_KEY
if not os.environ.get("NOVITA_API_KEY"):
print("Warning: NOVITA_API_KEY not set. Test will likely fail.")
def test_NOVITA_all():
text = "Apple Inc. was founded by Steve Jobs, Steve Wozniak, and Ronald Wayne in 1976. It is headquartered in Cupertino, California. The company designs, manufactures, and markets smartphones, personal computers, tablets, wearables, and accessories."
print("--- Testing NOVITA Provider Availability ---")
try:
provider = create_provider("NOVITA")
available = provider.is_available()
print(f"NOVITA Available: {available}")
if not available:
print("Error: NOVITA is not available. Check library installation or API key.")
return
except Exception as e:
print(f"Error checking provider: {e}")
return
print("\n--- Testing Entity Extraction ---")
try:
entities = extract_entities_llm(text, provider="NOVITA", model="deepseek/deepseek-v3.2")
print(f"Extracted {len(entities)} entities:")
pprint(entities)
except Exception as e:
print(f"Entity extraction failed: {e}")
print("\n--- Testing Relation Extraction ---")
try:
# Use a few entities for relation extraction
from semantica.semantic_extract.models import Entity
sample_entities = [
Entity(name="Apple Inc.", type="ORGANIZATION"),
Entity(name="Steve Jobs", type="PERSON")
]
relations = extract_relations_llm(text, entities=sample_entities, provider="NOVITA", model="deepseek/deepseek-v3.2")
print(f"Extracted {len(relations)} relations:")
pprint(relations)
except Exception as e:
print(f"Relation extraction failed: {e}")
print("\n--- Testing Triplet Extraction ---")
try:
triplets = extract_triplets_llm(text, provider="NOVITA", model="deepseek/deepseek-v3.2")
print(f"Extracted {len(triplets)} triplets:")
pprint(triplets)
except Exception as e:
print(f"Triplet extraction failed: {e}")
print("\n--- Testing Auto-Chunking ---")
long_text = " ".join([text] * 10) # Roughly 1000-1500 tokens
try:
entities_chunked = extract_entities_llm(
long_text,
provider="NOVITA",
model="deepseek/deepseek-v3.2",
max_text_length=200 # Force chunking
)
print(f"Extracted {len(entities_chunked)} entities from long text (chunked):")
# Just show count to avoid clutter
except Exception as e:
print(f"Chunked extraction failed: {e}")
if __name__ == "__main__":
test_NOVITA_all()