Source code for langchain_ibm.utils

from copy import deepcopy
from typing import TYPE_CHECKING, Any, Dict, Optional, Union

from ibm_watsonx_ai.foundation_models.schema import BaseSchema  # type: ignore
from pydantic import SecretStr

if TYPE_CHECKING:
    from langchain_ibm.toolkit import WatsonxTool


[docs] def check_for_attribute(value: SecretStr | None, key: str, env_key: str) -> None: if not value or not value.get_secret_value(): raise ValueError( f"Did not find {key}, please add an environment variable" f" `{env_key}` which contains it, or pass" f" `{key}` as a named parameter." )
[docs] def extract_params( kwargs: Dict[str, Any], default_params: Optional[Union[BaseSchema, Dict[str, Any]]] = None, ) -> Dict[str, Any]: if kwargs.get("params") is not None: params = kwargs.pop("params") elif default_params is not None: params = default_params else: params = None if isinstance(params, BaseSchema): params = params.to_dict() return params or {}
[docs] def extract_chat_params( kwargs: Dict[str, Any], default_params: Optional[Union[BaseSchema, Dict[str, Any]]] = None, ) -> Dict[str, Any]: if kwargs.get("params") is not None: params = kwargs.pop("params") check_duplicate_chat_params(params, kwargs) elif default_params is not None: params = deepcopy(default_params) else: params = None if isinstance(params, BaseSchema): params = params.to_dict() return params or {}
[docs] def check_duplicate_chat_params(params: dict, kwargs: dict) -> None: duplicate_keys = {k for k, v in kwargs.items() if v is not None and k in params} if duplicate_keys: raise ValueError( f"Duplicate parameters found in params and keyword arguments: " f"{list(duplicate_keys)}" )
[docs] def convert_to_watsonx_tool(tool: "WatsonxTool") -> dict: """Convert `WatsonxTool` to watsonx tool structure. Args: tool: `WatsonxTool` from `WatsonxToolkit` Example: .. code-block:: python from langchain_ibm import WatsonxToolkit watsonx_toolkit = WatsonxToolkit( url="https://us-south.ml.cloud.ibm.com", apikey="*****", ) weather_tool = watsonx_toolkit.get_tool("Weather") convert_to_watsonx_tool(weather_tool) # Return # { # "type": "function", # "function": { # "name": "Weather", # "description": "Find the weather for a city.", # "parameters": { # "type": "object", # "properties": { # "location": { # "title": "location", # "description": "Name of the location", # "type": "string", # }, # "country": { # "title": "country", # "description": "Name of the state or country", # "type": "string", # }, # }, # "required": ["location"], # }, # }, # } """ def parse_parameters(input_schema: dict | None) -> dict: if input_schema: parameters = deepcopy(input_schema) else: parameters = { "type": "object", "properties": { "input": { "description": "Input to be used when running tool.", "type": "string", }, }, "required": ["input"], } return parameters watsonx_tool = { "type": "function", "function": { "name": tool.name, "description": tool.description, "parameters": parse_parameters(tool.tool_input_schema), }, } return watsonx_tool