Method _call_model

internagent/mas/agents/base_agent.py:259–329

Protected method to call the language model with automatic retry logic. This is the primary interface for agents to interact with their language model. It provides robust error handling with exponential backoff retries, automatic selection between text and structure

(self,
                        prompt: str,
                        system_prompt: Optional[str] = None,
                        schema: Optional[Dict[str, Any]] = None,
                        temperature: Optional[float] = None)

        }
    
    async def _call_model(self,
                        prompt: str,
                        system_prompt: Optional[str] = None,
                        schema: Optional[Dict[str, Any]] = None,
                        temperature: Optional[float] = None) -> Union[str, Dict[str, Any]]:
        """
        Protected method to call the language model with automatic retry logic.

        This is the primary interface for agents to interact with their language model.
        It provides robust error handling with exponential backoff retries, automatic
        selection between text and structured (JSON) generation based on the schema
        parameter, and comprehensive logging of failures.

        All concrete agent implementations should use this method rather than calling
        the model directly to benefit from standardized error handling.

        Args:
            prompt (str): The main user prompt describing the task for the model.
                Should be clear, specific, and include all necessary context.
            system_prompt (Optional[str]): System-level instructions that guide the
                model's behavior and response style. If not provided, uses the agent's
                default system_prompt from configuration. Defaults to None.
            schema (Optional[Dict[str, Any]]): JSON Schema definition for structured
                output. When provided, enforces the model to return JSON matching this
                schema. When None, returns freeform text. Defaults to None.
            temperature (Optional[float]): Sampling temperature for model generation.
                Higher values (e.g., 0.8-1.0) increase creativity, lower values
                (e.g., 0.1-0.3) increase determinism. If None, uses model default.

        Returns:
            Union[str, Dict[str, Any]]: Model's response in one of two formats:
                - str: Freeform text response when schema is None
                - Dict[str, Any]: Structured JSON response when schema is provided

        Raises:
            AgentExecutionError: When model calls fail consistently after exhausting
                all retry attempts (max_retries). Contains details of the final error.

        Note:
            The method sleeps for 1 second between retry attempts to avoid hammering
            the API and potentially triggering rate limits. Consider this latency when
            designing time-sensitive operations.
        """
        system_prompt = system_prompt or self.system_prompt
        remaining_retries = self.max_retries
        
        while True:
            try:
                if schema:
                    return await self.model.generate_json(
                        prompt=prompt,
                        schema=schema,
                        system_prompt=system_prompt,
                        temperature=temperature
                    )
                else:
                    return await self.model.generate(
                        prompt=prompt,