########################################################################### No AI cache logic remains as per latest project requirements.
# Copyright (c) 2024, Oracle and/or its affiliates.
# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
#
# DISCLAIMER This is not an official Oracle application, It does not supported by Oracle Support.
#
# ai_repo.py
#
# @author: Andrew Gregory
#
# Supports Python 3.12 and above
#
# coding: utf-8
##########################################################################
import queue
from oci import config
from oci.auth.signers import InstancePrincipalsSecurityTokenSigner, SecurityTokenSigner
from oci.exceptions import ConfigFileNotFound, ServiceError
from oci.generative_ai import GenerativeAiClient
from oci.generative_ai_inference import GenerativeAiInferenceClient
from oci.generative_ai_inference.models import (
BaseChatRequest,
ChatDetails,
GenericChatRequest,
Message,
OnDemandServingMode,
TextContent,
)
from oci.signer import load_private_key_from_file
from oci_policy_analysis.common.logger import get_logger
# Global logger for this module
logger = get_logger(component='ai_repo')
[docs]
class AI:
"""AI Module for OCI Policy Analysis
Contains all of the available GenAI calls that can be made to obtain additional context.
Attributes:
genai_client: The OCI GenAI Client.
genai_inference_client: The OCI GenAI Inference Client
"""
def __init__(self):
"""Initialize OCI GenAI client and constants."""
logger.info('Initialized AI Module')
self.initialized = False
[docs]
def initialize_client(
self, use_instance_principal: bool, session_token: str | None = None, profile: str = 'DEFAULT'
) -> bool:
"""
Initialize OCI GenAI Client with authentication.
Args:
use_instance_principal (bool): Whether to use Instance Principal authentication.
session_token (str | None): Session token profile name for authentication.
profile (str): Profile name for authentication if not using instance principal or session token.
Returns:
bool: True if initialization is successful, False otherwise."""
try:
if use_instance_principal:
logger.debug('Using Instance Principal Authentication for AI')
self.signer = InstancePrincipalsSecurityTokenSigner()
self.genai_client = GenerativeAiClient(config={}, signer=self.signer)
self.genai_inference_client = GenerativeAiInferenceClient(config={}, signer=self.signer)
self.tenancy_ocid = self.signer.tenancy_id
self.region = self.signer.region
elif session_token:
logger.info('Attempt session auth')
self.config = config.from_file(profile_name=session_token)
token_file = self.config['security_token_file']
token = None
with open(token_file) as f:
token = f.read()
private_key = load_private_key_from_file(self.config['key_file'])
self.signer = SecurityTokenSigner(token, private_key)
self.genai_client = GenerativeAiClient(config={'region': self.config['region']}, signer=self.signer)
self.genai_inference_client = GenerativeAiInferenceClient(
config={'region': self.config['region']}, signer=self.signer
)
self.tenancy_ocid = self.config['tenancy']
logger.info('Success session auth')
else:
logger.debug(f'Using Profile Authentication for AI: {profile}')
self.config = config.from_file(profile_name=profile)
self.genai_client = GenerativeAiClient(self.config)
self.genai_inference_client = GenerativeAiInferenceClient(self.config)
self.tenancy_ocid = self.config['tenancy']
self.region = self.config['region']
logger.info(f'Set up GenAI and Inference Client for tenancy: {self.tenancy_ocid}')
# Set up base endpoint
self.base_endpoint = f'https://inference.generativeai.{self.region}.oci.oraclecloud.com'
self.initialized = True
return True
except (ConfigFileNotFound, Exception) as exc:
logger.fatal(f'Authentication failed: {exc}')
return False
[docs]
def update_config(self, model_ocid, endpoint, compartment_ocid):
"""
Update Model ID and Endpoint, reinitializing client if endpoint changes.
Args:
model_ocid (str): The model OCID to use.
endpoint (str): The endpoint URL to use.
compartment_ocid (str): The compartment OCID for requests."""
logger.info(
f'Updating AI config: Model OCID:{model_ocid}, Endpoint:{endpoint}, Compartment: {compartment_ocid}'
)
self.model_ocid = model_ocid
self.endpoint = endpoint
self.compartment_ocid = compartment_ocid
def _create_chat_request(self, prompt) -> ChatDetails:
"""Create a Chat Request for the given prompt.
Args:
prompt (str): The prompt to send to the AI model.
Returns:
ChatDetails: The constructed chat request details.
"""
chat_detail = ChatDetails()
chat_detail.serving_mode = OnDemandServingMode(model_id=self.model_ocid)
content = TextContent()
content.text = prompt
chat_request = GenericChatRequest()
chat_request.api_format = BaseChatRequest.API_FORMAT_GENERIC
chat_request.messages = [Message(role='USER', content=[content])]
# chat_request.max_tokens = 1500
chat_request.temperature = 0
chat_request.top_p = 0.25
# chat_request.top_k = 0
# TODO: Change the parameters above based on chosen model. Need to test with multiple models to determine best defaults and if any parameters should be locked to specific values for certain models.
chat_detail.chat_request = chat_request
chat_detail.compartment_id = self.compartment_ocid
logger.info(f'Created Chat Request with prompt: {prompt}')
logger.debug(f'Created Chat: {chat_detail}')
return chat_detail
[docs]
def list_models(self) -> list[dict]:
"""
List available models using GenerativeAiClient.list_models.
Returns:
list[dict]: A list of available GenAI models with their details.
"""
logger.info('Listing available models')
try:
# Try to list models from tenancy
response = self.genai_client.list_models(compartment_id=self.tenancy_ocid)
if (
response is None
or getattr(response, 'data', None) is None
or getattr(response.data, 'items', None) is None
):
logger.error('list_models response or response.data or response.data.items is None')
return []
models = [
{
'Model Name': model.display_name or 'Unknown',
'Model OCID': model.id,
'Capabilities': ', '.join(model.capabilities) if model.capabilities else 'N/A',
'Lifecycle State': model.lifecycle_state or 'N/A',
'Creation Date': model.time_created.isoformat() if model.time_created else 'N/A',
}
for model in response.data.items
]
logger.info('Retrieved %d models from list_models', len(models))
return models
except ServiceError as e:
logger.error('Service error listing models: %s', e)
raise
except Exception as e:
logger.error('Error listing models: %s', e)
raise
[docs]
async def analyze_policy_statement(
self,
policy_text: str,
format: str = 'Markdown',
queue: queue.Queue | None = None,
additional_instruction: str = '',
):
"""
Analyze a policy statement using the GenAI Inference Client.
Args:
policy_text (str): The policy statement text to analyze.
format (str): The result format to use. "Markdown" means the GenAI should return markdown, "Text" means plain text using newlines for paragraphs and no bullet points, no markdown, no HTML.
queue (queue.Queue|None): Optional queue to put the result into for async calls.
additional_instruction (str): Additional instructions to include in the prompt.
Returns:
str: The analysis result from the AI model.
"""
logger.info(f'Analyzing policy statement: {policy_text} (Format={format})')
if additional_instruction == 'TEST':
if format.upper() == 'TEXT':
prompt = 'What is the meaning of life? Return witty response quickly, in plain text only, no markdown, no bullets, no lists, newlines as paragraph markers.'
else:
prompt = 'What is the meaning of life? Return witty response quickly. Use Strict Markdown only.'
else:
if format.upper() == 'TEXT':
prompt = (
f"Describe OCI Policy permission '{policy_text}' in detail using only plain text with clear, well-separated paragraphs. "
'Do not use any markdown, bullet points, lists, HTML, or formatting: just pure readable text, and use blank lines (double newlines) for paragraph breaks. '
'If you include any reference link, write the full https://... on its own line. '
f'{additional_instruction}'
)
else:
prompt = (
f"Describe OCI Policy permission '{policy_text}' in detail with clear markdown sections. "
'Always return strict markdown with #### for sections, one level of un-ordered lists, no new lines except for semantic breaks, and documentation link if possible. '
f'{additional_instruction}'
)
chat_detail = self._create_chat_request(prompt=prompt)
try:
response = self.genai_inference_client.chat(chat_detail)
if (
response is not None
and getattr(response, 'data', None) is not None
and getattr(response.data, 'chat_response', None) is not None
and getattr(response.data.chat_response, 'choices', None)
and len(response.data.chat_response.choices) > 0
and getattr(response.data.chat_response.choices[0], 'message', None) is not None
and getattr(response.data.chat_response.choices[0].message, 'content', None) is not None
):
content = response.data.chat_response.choices[0].message.content
# Always treat result as string for user display.
result = content if isinstance(content, str) else str(content)
else:
logger.error('AI chat response was None or incomplete (missing required attributes)')
result = 'Error: GenAI service response incomplete or None'
except ServiceError as e:
logger.error(f'Ai Service error: {e}')
result = f'Error: GenAI service error ({e.status})'
except Exception as e:
logger.error(f'Unexpected error: {e}')
result = f'Error: {str(e)}'
if queue is not None:
queue.put(result)
else:
return result
# test_ai_call method has been removed per current requirements.