Skip to content

Cloud

CloudAgent

Bases: BaseAgent

Source code in Docs2KG/agents/cloud.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class CloudAgent(BaseAgent):
    def __init__(self, name: str):
        """
        Initialize CloudAgent with model name and optional API key.

        Args:
            name: Name of the model to use (e.g., 'gpt-4')
        """
        super().__init__(name)
        self.client = self._init_openai_client()

    def _init_openai_client(self) -> OpenAI:
        """
        Initialize OpenAI client with API key from either:
        1. Explicitly passed api_key parameter
        2. Environment variable OPENAI_API_KEY
        3. .env file
        """
        try:
            # Initialize client with config
            client = OpenAI(
                api_key=PROJECT_CONFIG.openai.api_key.get_secret_value(),
                base_url=PROJECT_CONFIG.openai.api_base,
                timeout=PROJECT_CONFIG.openai.timeout,
                max_retries=PROJECT_CONFIG.openai.max_retries,
            )

            logger.info(f"Successfully initialized OpenAI client for model {self.name}")
            return client

        except Exception as e:
            logger.error(f"Failed to initialize OpenAI client: {str(e)}")
            raise

    def process(self, input_data: Any) -> Any:
        """
        Process input using the OpenAI client.

        Args:
            input_data: The input to be processed by the model

        Returns:
            Dict containing the model response and metadata
        """
        logger.info(f"Processing input with OpenAI: {input_data}")

        try:
            # Create chat completion with proper error handling
            response = self.client.chat.completions.create(
                model=self.name,
                messages=[{"role": "user", "content": str(input_data)}],
                max_tokens=PROJECT_CONFIG.openai.max_tokens,
                temperature=PROJECT_CONFIG.openai.temperature,
            )

            return {
                "model": self.name,
                "input": input_data,
                "status": "processed",
                "response": response.choices[0].message.content,
                "usage": response.usage.dict() if response.usage else None,
            }

        except Exception as e:
            logger.error(f"Error processing input with OpenAI: {str(e)}")
            raise

__init__(name)

Initialize CloudAgent with model name and optional API key.

Parameters:

Name Type Description Default
name str

Name of the model to use (e.g., 'gpt-4')

 required
Source code in Docs2KG/agents/cloud.py 
11
12
13
14
15
16
17
18
19
def __init__(self, name: str):
    """
    Initialize CloudAgent with model name and optional API key.

    Args:
        name: Name of the model to use (e.g., 'gpt-4')
    """
    super().__init__(name)
    self.client = self._init_openai_client()

process(input_data)

Process input using the OpenAI client.

Parameters:

Name Type Description Default
input_data Any

The input to be processed by the model

 required

Returns:

Type Description
Any

Dict containing the model response and metadata
Source code in Docs2KG/agents/cloud.py 
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def process(self, input_data: Any) -> Any:
    """
    Process input using the OpenAI client.

    Args:
        input_data: The input to be processed by the model

    Returns:
        Dict containing the model response and metadata
    """
    logger.info(f"Processing input with OpenAI: {input_data}")

    try:
        # Create chat completion with proper error handling
        response = self.client.chat.completions.create(
            model=self.name,
            messages=[{"role": "user", "content": str(input_data)}],
            max_tokens=PROJECT_CONFIG.openai.max_tokens,
            temperature=PROJECT_CONFIG.openai.temperature,
        )

        return {
            "model": self.name,
            "input": input_data,
            "status": "processed",
            "response": response.choices[0].message.content,
            "usage": response.usage.dict() if response.usage else None,
        }

    except Exception as e:
        logger.error(f"Error processing input with OpenAI: {str(e)}")
        raise