Source code for orkgnlp.deepresearch.deepresearch

# -*- coding: utf-8 -*-
import warnings
from typing import Any

from orkgnlp.common.config import orkgnlp_context
from orkgnlp.common.service.base import (
    ORKGNLPBaseDecoder,
    ORKGNLPBaseEncoder,
    ORKGNLPBaseRunner,
    ORKGNLPBaseService,
)
from orkgnlp.deepresearch.encoder import DeepResearchEncoder
from orkgnlp.deepresearch.runner import DeepResearchRunner




class DeepResearch(ORKGNLPBaseService):
    SERVICE_NAME = "deep-research"

    def __init__(
        self,
        openai_api_key: str = None,
        firecrawl_api_key: str = None,
        research_provider: str = "orkg",
        openai_endpoint: str = "https://api.openai.com/v1",
        custom_model: str = "o3-mini",
        *args: Any,
        **kwargs: Any
    ):
        super().__init__(self.SERVICE_NAME, *args, **kwargs)
        warnings.warn(
            "Currently ORKG-NLP Deep Research is not operational inside the Jupyter Notebooks.",
            RuntimeWarning,
        )

        encoder: ORKGNLPBaseEncoder = DeepResearchEncoder(
            openai_api_key=openai_api_key,
            firecrawl_api_key=firecrawl_api_key,
            research_provider=research_provider,
            openai_endpoint=openai_endpoint,
            custom_model=custom_model,
        )
        runner: ORKGNLPBaseRunner = DeepResearchRunner()
        decoder: ORKGNLPBaseDecoder = ORKGNLPBaseDecoder()
        self._register_pipeline("main", encoder, runner, decoder)



    def __call__(
        self,
        query: str,
        depth: int = 2,
        breadth: int = 4,
        report_type: str = "report",
    ):
        return self._run(
            raw_input=query,
            depth=depth,
            breadth=breadth,
            report_type=report_type,
            pipline_executor_name="main",
        )




orkgnlp_context.get("SERVICE_MAP")[DeepResearch.SERVICE_NAME] = DeepResearch