# -*- coding: utf-8 -*-
# encoder.py
from typing import Any, Dict, Tuple
from overrides import overrides
from orkgnlp.common.service.base import ORKGNLPBaseEncoder
from orkgnlp.deepresearch.client import FirecrawlApp, ORKGAskApp
from orkgnlp.deepresearch.config import ModelConfig
[docs]
class DeepResearchEncoder(ORKGNLPBaseEncoder):
def __init__(
self,
openai_api_key: str,
firecrawl_api_key: str = None,
research_provider: str = "orkg",
openai_endpoint: str = "https://api.openai.com/v1",
custom_model: str = "o3-mini",
):
super().__init__()
self.openai_api_key = openai_api_key
self.firecrawl_api_key = firecrawl_api_key
self.research_provider = research_provider
self.openai_endpoint = openai_endpoint
self.custom_model = custom_model
[docs]
@overrides
def encode(self, raw_input: Any, **kwargs: Any) -> Tuple[Any, Dict[str, Any]]:
if not raw_input or raw_input.strip().lower() == "null":
raise ValueError("Please enter a query you would like to research")
if not self.openai_api_key:
raise ValueError("Missing required field: openai_api_key")
config = ModelConfig(
api_key=self.openai_api_key,
base_url=self.openai_endpoint,
custom_model=self.custom_model,
)
if self.research_provider == "orkg":
search_client = ORKGAskApp()
elif self.research_provider == "firecrawl":
if not self.firecrawl_api_key:
raise ValueError(
"Missing required field: firecrawl_api_key (required when research_provider='firecrawl')"
)
search_client = FirecrawlApp(
api_key=self.firecrawl_api_key, base_url="https://api.firecrawl.dev/v1"
)
else:
raise ValueError(
f"Unknown research_provider='{self.research_provider}'. Expected 'firecrawl' or 'orkg'."
)
kwargs["model_config"] = config
kwargs["search_client"] = search_client
return (raw_input,), kwargs