aidputils.agents.toolkit.custom_oci_generative_ai

class aidputils.agents.toolkit.custom_oci_generative_ai.OCIUtils

Bases: object

Utility functions for OCI Generative AI integration.

static is_pydantic_class(obj: Any) → bool

Check if an object is a Pydantic BaseModel subclass.

static remove_signature_from_tool_description(name: str, description: str) → str

Remove the tool signature and Args section from a tool description.

The signature is typically prefixed to the description and followed

by an Args section.

static convert_oci_tool_call_to_langchain(tool_call: Any) → langchain_core.messages.ToolCall

Convert an OCI tool call to a LangChain ToolCall.

class aidputils.agents.toolkit.custom_oci_generative_ai.Provider

Bases: ABC

Abstract base class for OCI Generative AI providers.

abstract property stop_sequence_key: str

Return the stop sequence key for the provider.

abstractmethod chat_response_to_text(response: Any) → str

Extract chat text from a provider’s response.

abstractmethod chat_stream_to_text(event_data: Dict) → str

Extract chat text from a streaming event.

abstractmethod is_chat_stream_end(event_data: Dict) → bool

Determine if the chat stream event marks the end of a stream.

abstractmethod chat_generation_info(response: Any) → Dict[str, Any]

Extract generation metadata from a provider’s response.

abstractmethod chat_stream_generation_info(event_data: Dict) → Dict[str, Any]

Extract generation metadata from a chat stream event.

abstractmethod chat_tool_calls(response: Any) → List[Any]

Extract tool calls from a provider’s response.

abstractmethod chat_stream_tool_calls(event_data: Dict) → List[Any]

Extract tool calls from a streaming event.

abstractmethod format_response_tool_calls(tool_calls: List[Any]) → List[Any]

Format response tool calls into LangChain’s expected structure.

abstractmethod format_stream_tool_calls(tool_calls: List[Any]) → List[Any]

Format stream tool calls into LangChain’s expected structure.

abstractmethod get_role(message: langchain_core.messages.BaseMessage) → str

Map a LangChain message to the provider’s role representation.

abstractmethod messages_to_oci_params(messages: Any, **kwargs: Any) → Dict[str, Any]

Convert LangChain messages to OCI API parameters.

abstractmethod convert_to_oci_tool(tool: Dict[str, Any] | Type[pydantic.BaseModel] | Callable | langchain_core.tools.BaseTool) → Dict[str, Any]

Convert a tool definition into the provider-specific OCI tool format.

abstractmethod process_tool_choice(tool_choice: dict | str | Literal['auto', 'none', 'required', 'any'] | bool | None) → Any | None

Process tool choice parameter for the provider.

abstractmethod process_stream_tool_calls(event_data: Dict, tool_call_ids: Set[str]) → List[langchain_core.messages.tool.ToolCallChunk]

Process streaming tool calls from event data into chunks.

class aidputils.agents.toolkit.custom_oci_generative_ai.OCIAIMessageChunk(*args: Any, **kwargs: Any)

Bases: AIMessageChunk

init_tool_calls() → Self

Initialize tool calls from tool call chunks.

Returns:

The values with tool calls initialized.

Raises:

ValueError – If the tool call chunks are malformed.

class aidputils.agents.toolkit.custom_oci_generative_ai.CohereProvider

Bases: Provider

Provider implementation for Cohere.

stop_sequence_key: str = 'stop_sequences'

chat_response_to_text(response: Any) → str

Extract text from a Cohere chat response.

chat_stream_to_text(event_data: Dict) → str

Extract text from a Cohere chat stream event.

is_chat_stream_end(event_data: Dict) → bool

Determine if the Cohere stream event indicates the end.

chat_generation_info(response: Any) → Dict[str, Any]

Extract generation information from a Cohere chat response.

chat_stream_generation_info(event_data: Dict) → Dict[str, Any]

Extract generation info from a Cohere chat stream event.

chat_tool_calls(response: Any) → List[Any]

Retrieve tool calls from a Cohere chat response.

chat_stream_tool_calls(event_data: Dict) → List[Any]

Retrieve tool calls from Cohere stream event data.

format_response_tool_calls(tool_calls: List[Any] | None = None) → List[Dict]

Formats a OCI GenAI API Cohere response into the tool call format used in Langchain.

format_stream_tool_calls(tool_calls: List[Any]) → List[Dict]

Formats a OCI GenAI API Cohere stream response into the tool call format used in Langchain.

get_role(message: langchain_core.messages.BaseMessage) → str

Map a LangChain message to Cohere’s role representation.

messages_to_oci_params(messages: Sequence[langchain_core.messages.ChatMessage], **kwargs: Any) → Dict[str, Any]

Convert LangChain messages to OCI parameters for Cohere.

This includes conversion of chat history and tool call results.

convert_to_oci_tool(tool: Dict[str, Any] | Type[pydantic.BaseModel] | Callable | langchain_core.tools.BaseTool) → Dict[str, Any]

Convert a tool definition to an OCI tool for Cohere.

Supports BaseTool instances, JSON schema dictionaries,

or Pydantic models/callables.

process_tool_choice(tool_choice: dict | str | Literal['auto', 'none', 'required', 'any'] | bool | None) → Any | None

Cohere does not support tool choices.

process_stream_tool_calls(event_data: Dict, tool_call_ids: Set[str]) → List[langchain_core.messages.tool.ToolCallChunk]

Process Cohere stream tool calls and return them as ToolCallChunk objects.

Parameters:

• event_data – The event data from the stream

• tool_call_ids – Set of existing tool call IDs for index tracking

Returns:

List of ToolCallChunk objects

class aidputils.agents.toolkit.custom_oci_generative_ai.GenericProvider

Bases: Provider

Provider for models using generic API spec.

stop_sequence_key: str = 'stop'

chat_response_to_text(response: Any) → str

Extract text from Meta chat response.

chat_stream_to_text(event_data: Dict) → str

Extract text from Meta chat stream event.

is_chat_stream_end(event_data: Dict) → bool

Determine if Meta chat stream event indicates the end.

chat_generation_info(response: Any) → Dict[str, Any]

Extract generation metadata from Meta chat response.

chat_stream_generation_info(event_data: Dict) → Dict[str, Any]

Extract generation metadata from Meta chat stream event.

chat_tool_calls(response: Any) → List[Any]

Retrieve tool calls from Meta chat response.

chat_stream_tool_calls(event_data: Dict) → List[Any]

Retrieve tool calls from Meta stream event.

format_response_tool_calls(tool_calls: List[Any]) → List[Dict]

Formats a OCI GenAI API Meta response into the tool call format used in Langchain.

format_stream_tool_calls(tool_calls: List[Any] | None = None) → List[Dict]

Formats a OCI GenAI API Meta stream response into the tool call format used in Langchain.

get_role(message: langchain_core.messages.BaseMessage) → str

Map a LangChain message to Meta’s role representation.

messages_to_oci_params(messages: List[langchain_core.messages.BaseMessage], **kwargs: Any) → Dict[str, Any]

Convert LangChain messages to OCI chat parameters.

Parameters:

• messages – List of LangChain BaseMessage objects

• **kwargs – Additional keyword arguments

Returns:

Dict containing OCI chat parameters

Raises:

ValueError – If message content is invalid

convert_to_oci_tool(tool: Type[pydantic.BaseModel] | Callable | langchain_core.tools.BaseTool) → Dict[str, Any]

Convert a BaseTool instance, TypedDict or BaseModel type to a OCI tool in Meta’s format.

Parameters:

tool – The tool to convert, can be a BaseTool instance, TypedDict, or BaseModel type.

Returns:

Dict containing the tool definition in Meta’s format.

Raises:

ValueError – If the tool type is not supported.

process_tool_choice(tool_choice: dict | str | Literal['auto', 'none', 'required', 'any'] | bool | None) → Any | None

Process tool choice for Meta provider.

Parameters:

tool_choice –

Which tool to require the model to call. Options are: - str of the form “<<tool_name>>”: calls <<tool_name>> tool. - “auto”: automatically selects a tool (including no tool). - “none”: does not call a tool. - “any” or “required” or True: force at least one tool to be called. - dict of the form

{“type”: “function”, “function”: {“name”: <<tool_name>>}}:

calls <<tool_name>> tool. - False or None: no effect, default Meta behavior.

Returns:

Meta-specific tool choice object.

Raises:

ValueError – If tool_choice type is not recognized.

process_stream_tool_calls(event_data: Dict, tool_call_ids: Set[str]) → List[langchain_core.messages.tool.ToolCallChunk]

Process Meta stream tool calls and convert them to ToolCallChunks.

Parameters:

• event_data – The event data from the stream

• tool_call_ids – Set of existing tool call IDs for index tracking

Returns:

List of ToolCallChunk objects

class aidputils.agents.toolkit.custom_oci_generative_ai.MetaProvider

Bases: GenericProvider

Provider for Meta models. This provider is for backward compatibility.

class aidputils.agents.toolkit.custom_oci_generative_ai.ChatOCIGenAI(*args: Any, **kwargs: Any)

Bases: BaseChatModel, OCIGenAIBase

ChatOCIGenAI chat model integration.

Setup:

Install langchain-oci and the oci sdk.

pip install -U langchain-oci oci

Key init args — completion params:

model_id: str

Id of the OCIGenAI chat model to use, e.g., cohere.command-r-16k.

is_stream: bool

Whether to stream back partial progress

model_kwargs: Optional[Dict]

Keyword arguments to pass to the specific model used, e.g., temperature, max_tokens.

Key init args — client params:

service_endpoint: str

The endpoint URL for the OCIGenAI service, e.g., https://inference.generativeai.us-chicago-1.oci.oraclecloud.com.

compartment_id: str

The compartment OCID.

auth_type: str

The authentication type to use, e.g., API_KEY (default), SECURITY_TOKEN, INSTANCE_PRINCIPAL, RESOURCE_PRINCIPAL.

auth_profile: Optional[str]

The name of the profile in ~/.oci/config, if not specified , DEFAULT will be used.

auth_file_location: Optional[str]

Path to the config file, If not specified, ~/.oci/config will be used.

provider: str

Provider name of the model. Default to None, will try to be derived from the model_id otherwise, requires user input.

See full list of supported init args and their descriptions in the params section.

Instantiate:

from langchain_oci.chat_models import ChatOCIGenAI

chat = ChatOCIGenAI(
    model_id="cohere.command-r-16k",
    service_endpoint="https://inference.generativeai.us-chicago-1.oci.oraclecloud.com",
    compartment_id="MY_OCID",
    model_kwargs={"temperature": 0.7, "max_tokens": 500},
)

Invoke:

Stream:

for r in chat.stream(messages):

print(r.content, end=””, flush=True)

Response metadata

response = chat.invoke(messages) print(response.response_metadata)

bind_tools(tools: ~typing.Sequence[~typing.Dict[str, ~typing.Any] | ~typing.Type[pydantic.BaseModel] | ~typing.Callable | langchain_core.tools.BaseTool], *, tool_choice: dict | str | ~typing.Literal['auto', 'none', 'required', 'any'] | bool | None = None, **kwargs: ~typing.Any) -> langchain_core.runnables.Runnable.(langchain_core.language_models.LanguageModelInput, langchain_core.messages.BaseMessage)

Bind tool-like objects to this chat model.

Assumes model is compatible with Meta’s tool-calling API.

Parameters:

• tools – A list of tool definitions to bind to this chat model. Can be a dictionary, pydantic model, or callable. Pydantic models and callables will be automatically converted to their schema dictionary representation.

• tool_choice –

  Which tool to require the model to call. Options are: - str of the form “<<tool_name>>”: calls <<tool_name>> tool. - “auto”: automatically selects a tool (including no tool). - “none”: does not call a tool. - “any” or “required” or True: force at least one tool to be called. - dict of the form

  {“type”: “function”, “function”: {“name”: <<tool_name>>}}:

  calls <<tool_name>> tool. - False or None: no effect, default Meta behavior.

• kwargs – Any additional parameters are passed directly to bind().

with_structured_output(schema: ~typing.Dict | ~typing.Type[pydantic.BaseModel] | None = None, *, method: ~typing.Literal['function_calling', 'json_schema', 'json_mode'] = 'function_calling', include_raw: bool = False, **kwargs: ~typing.Any) -> langchain_core.runnables.Runnable.(langchain_core.language_models.LanguageModelInput, typing.Union[typing.Dict, pydantic.BaseModel])

Model wrapper that returns outputs formatted to match the given schema.

Parameters:

• schema – The output schema as a dict or a Pydantic class. If a Pydantic class then the model output will be an object of that class. If a dict then the model output will be a dict. With a Pydantic class the returned attributes will be validated, whereas with a dict they will not be. If method is “function_calling” and schema is a dict, then the dict must match the OCI Generative AI function-calling spec.

• method – The method for steering model generation, either “function_calling” or “json_mode” or “json_schema. If “function_calling” then the schema will be converted to an OCI function and the returned model will make use of the function-calling API. If “json_mode” then Cohere’s JSON mode will be used. Note that if using “json_mode” then you must include instructions for formatting the output into the desired schema into the model call. If “json_schema” then it allows the user to pass a json schema (or pydantic) to the model for structured output. This is the default method.

• include_raw – If False then only the parsed structured output is returned. If an error occurs during model output parsing it will be raised. If True then both the raw model response (a BaseMessage) and the parsed model response will be returned. If an error occurs during output parsing it will be caught and returned as well. The final output is always a dict with keys “raw”, “parsed”, and “parsing_error”.

Returns:

If include_raw is True then a dict with keys:

raw: BaseMessage parsed: Optional[_DictOrPydantic] parsing_error: Optional[BaseException]

If include_raw is False then just _DictOrPydantic is returned, where _DictOrPydantic depends on the schema:

If schema is a Pydantic class then _DictOrPydantic is the Pydantic

class.

If schema is a dict then _DictOrPydantic is a dict.

Return type:

A Runnable that takes any ChatModel input and returns as output

emit_usage_data(usage)