github/langchain
1
0
Fork 0
mirror of https://github.com/hwchase17/langchain.git synced 2025-07-03 11:47:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

ibm: move to external repo (#24208)

Browse Source
This commit is contained in:
Erick Friis 2024-07-12 23:14:24 +02:00 committed by GitHub
parent 02e71cebed
commit e8ee781a42
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
25 changed files with 2 additions and 3842 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
libs/partners/ibm/.gitignore vendored
View File

@ -1 +0,0 @@
__pycache__

21
libs/partners/ibm/LICENSE
View File

@ -1,21 +0,0 @@
MIT License
Copyright (c) 2023 LangChain, Inc.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

58
libs/partners/ibm/Makefile
View File

@ -1,58 +0,0 @@
.PHONY: all format lint test tests integration_tests docker_tests help extended_tests
# Default target executed when no arguments are given to make.
all: help
# Define a variable for the test file path.
TEST_FILE ?= tests/unit_tests/
integration_test integration_tests: TEST_FILE=tests/integration_tests/
test tests integration_test integration_tests:
poetry run pytest $(TEST_FILE)
######################
# LINTING AND FORMATTING
######################
# Define a variable for Python and notebook files.
PYTHON_FILES=.
MYPY_CACHE=.mypy_cache
lint format: PYTHON_FILES=.
lint_diff format_diff: PYTHON_FILES=$(shell git diff --relative=libs/partners/ibm --name-only --diff-filter=d master | grep -E '\.py$$|\.ipynb$$')
lint_package: PYTHON_FILES=langchain_ibm
lint_tests: PYTHON_FILES=tests
lint_tests: MYPY_CACHE=.mypy_cache_test
lint lint_diff lint_package lint_tests:
poetry run ruff check .
poetry run ruff format $(PYTHON_FILES) --diff
poetry run ruff check --select I $(PYTHON_FILES)
mkdir $(MYPY_CACHE); poetry run mypy $(PYTHON_FILES) --cache-dir $(MYPY_CACHE)
format format_diff:
poetry run ruff format $(PYTHON_FILES)
poetry run ruff check --select I --fix $(PYTHON_FILES)
spell_check:
poetry run codespell --toml pyproject.toml
spell_fix:
poetry run codespell --toml pyproject.toml -w
check_imports: $(shell find langchain_ibm -name '*.py')
poetry run python ./scripts/check_imports.py $^
######################
# HELP
######################
help:
@echo '----'
@echo 'check_imports - check imports'
@echo 'format - run code formatters'
@echo 'lint - run linters'
@echo 'test - run unit tests'
@echo 'tests - run unit tests'
@echo 'test TEST_FILE=<test_file> - run all tests in file'

145
libs/partners/ibm/README.md
View File

@ -1,144 +1,3 @@
# langchain-ibm This package has moved!
This package provides the integration between LangChain and IBM watsonx.ai through the `ibm-watsonx-ai` SDK. https://github.com/langchain-ai/langchain-ibm/tree/main/libs/ibm
## Installation
To use the `langchain-ibm` package, follow these installation steps:
```bash
pip install langchain-ibm
```
## Usage
### Setting up
To use IBM's models, you must have an IBM Cloud user API key. Here's how to obtain and set up your API key:
1. **Obtain an API Key:** For more details on how to create and manage an API key, refer to IBM's [documentation](https://cloud.ibm.com/docs/account?topic=account-userapikey&interface=ui).
2. **Set the API Key as an Environment Variable:** For security reasons, it's recommended to not hard-code your API key directly in your scripts. Instead, set it up as an environment variable. You can use the following code to prompt for the API key and set it as an environment variable:
```python
import os
from getpass import getpass
watsonx_api_key = getpass()
os.environ["WATSONX_APIKEY"] = watsonx_api_key
```
In alternative, you can set the environment variable in your terminal.
- **Linux/macOS:** Open your terminal and execute the following command:
```bash
export WATSONX_APIKEY='your_ibm_api_key'
```
To make this environment variable persistent across terminal sessions, add the above line to your `~/.bashrc`, `~/.bash_profile`, or `~/.zshrc` file.
- **Windows:** For Command Prompt, use:
```cmd
set WATSONX_APIKEY=your_ibm_api_key
```
### Loading the model
You might need to adjust model parameters for different models or tasks. For more details on the parameters, refer to IBM's [documentation](https://ibm.github.io/watsonx-ai-python-sdk/fm_model.html#metanames.GenTextParamsMetaNames).
```python
parameters = {
"decoding_method": "sample",
"max_new_tokens": 100,
"min_new_tokens": 1,
"temperature": 0.5,
"top_k": 50,
"top_p": 1,
}
```
Initialize the WatsonxLLM class with the previously set parameters.
```python
from langchain_ibm import WatsonxLLM
watsonx_llm = WatsonxLLM(
model_id="PASTE THE CHOSEN MODEL_ID HERE",
url="PASTE YOUR URL HERE",
project_id="PASTE YOUR PROJECT_ID HERE",
params=parameters,
)
```
**Note:**
- You must provide a `project_id` or `space_id`. For more information refer to IBM's [documentation](https://www.ibm.com/docs/en/watsonx-as-a-service?topic=projects).
- Depending on the region of your provisioned service instance, use one of the urls described [here](https://ibm.github.io/watsonx-ai-python-sdk/setup_cloud.html#authentication).
- You need to specify the model you want to use for inferencing through `model_id`. You can find the list of available models [here](https://ibm.github.io/watsonx-ai-python-sdk/fm_model.html#ibm_watsonx_ai.foundation_models.utils.enums.ModelTypes).
Alternatively you can use Cloud Pak for Data credentials. For more details, refer to IBM's [documentation](https://ibm.github.io/watsonx-ai-python-sdk/setup_cpd.html).
```python
watsonx_llm = WatsonxLLM(
model_id="ibm/granite-13b-instruct-v2",
url="PASTE YOUR URL HERE",
username="PASTE YOUR USERNAME HERE",
password="PASTE YOUR PASSWORD HERE",
instance_id="openshift",
version="4.8",
project_id="PASTE YOUR PROJECT_ID HERE",
params=parameters,
)
```
### Create a Chain
Create `PromptTemplate` objects which will be responsible for creating a random question.
```python
from langchain.prompts import PromptTemplate
template = "Generate a random question about {topic}: Question: "
prompt = PromptTemplate.from_template(template)
```
Provide a topic and run the LLMChain.
```python
from langchain.chains import LLMChain
llm_chain = LLMChain(prompt=prompt, llm=watsonx_llm)
response = llm_chain.invoke("dog")
print(response)
```
### Calling the Model Directly
To obtain completions, you can call the model directly using a string prompt.
```python
# Calling a single prompt
response = watsonx_llm.invoke("Who is man's best friend?")
print(response)
```
```python
# Calling multiple prompts
response = watsonx_llm.generate(
[
"The fastest dog in the world?",
"Describe your chosen dog breed",
]
)
print(response)
```
### Streaming the Model output
You can stream the model output.
```python
for chunk in watsonx_llm.stream(
"Describe your favorite breed of dog and why it is your favorite."
):
print(chunk, end="")
```

5
libs/partners/ibm/langchain_ibm/__init__.py
View File

@ -1,5 +0,0 @@
from langchain_ibm.chat_models import ChatWatsonx
from langchain_ibm.embeddings import WatsonxEmbeddings
from langchain_ibm.llms import WatsonxLLM
__all__ = ["WatsonxLLM", "WatsonxEmbeddings", "ChatWatsonx"]

915
libs/partners/ibm/langchain_ibm/chat_models.py
View File

@ -1,915 +0,0 @@
import json
import logging
import os
import re
from operator import itemgetter
from typing import (
Any,
Callable,
Dict,
Iterator,
List,
Literal,
Mapping,
Optional,
Sequence,
Tuple,
Type,
TypedDict,
Union,
cast,
)
from ibm_watsonx_ai import Credentials # type: ignore
from ibm_watsonx_ai.foundation_models import ModelInference # type: ignore
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models import LanguageModelInput
from langchain_core.language_models.chat_models import (
BaseChatModel,
LangSmithParams,
generate_from_stream,
)
from langchain_core.messages import (
AIMessage,
AIMessageChunk,
BaseMessage,
BaseMessageChunk,
ChatMessage,
ChatMessageChunk,
FunctionMessage,
FunctionMessageChunk,
HumanMessage,
HumanMessageChunk,
SystemMessage,
SystemMessageChunk,
ToolCallChunk,
ToolMessage,
ToolMessageChunk,
convert_to_messages,
)
from langchain_core.messages.tool import tool_call_chunk as create_tool_call_chunk
from langchain_core.output_parsers import JsonOutputParser, PydanticOutputParser
from langchain_core.output_parsers.base import OutputParserLike
from langchain_core.output_parsers.openai_tools import (
JsonOutputKeyToolsParser,
PydanticToolsParser,
make_invalid_tool_call,
parse_tool_call,
)
from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, ChatResult
from langchain_core.prompt_values import ChatPromptValue
from langchain_core.pydantic_v1 import BaseModel, Field, SecretStr, root_validator
from langchain_core.runnables import Runnable, RunnableMap, RunnablePassthrough
from langchain_core.tools import BaseTool
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env
from langchain_core.utils.function_calling import (
convert_to_openai_function,
convert_to_openai_tool,
)
logger = logging.getLogger(__name__)
def _convert_dict_to_message(_dict: Mapping[str, Any]) -> BaseMessage:
"""Convert a dictionary to a LangChain message.
Args:
_dict: The dictionary.
Returns:
The LangChain message.
"""
role = _dict.get("role")
if role == "user":
return HumanMessage(content=_dict.get("generated_text", ""))
else:
additional_kwargs: Dict = {}
tool_calls = []
invalid_tool_calls = []
try:
content = ""
raw_tool_calls = _dict.get("generated_text")
if raw_tool_calls:
json_parts = re.split(r"\n\n(?:<blank line>\n\n)?", raw_tool_calls)
parsed_raw_tool_calls = [
json.loads(part) for part in json_parts if part.strip()
]
additional_kwargs["tool_calls"] = parsed_raw_tool_calls
additional_kwargs["function_call"] = dict(parsed_raw_tool_calls)
for obj in parsed_raw_tool_calls:
b = json.dumps(obj["function"]["arguments"])
obj["function"]["arguments"] = b
for raw_tool_call in parsed_raw_tool_calls:
try:
raw_tool_call["id"] = "None"
tool_calls.append(
parse_tool_call(raw_tool_call, return_id=True)
)
except Exception as e:
invalid_tool_calls.append(
dict(make_invalid_tool_call(raw_tool_call, str(e)))
)
except: # noqa: E722
content = _dict.get("generated_text", "") or ""
return AIMessage(
content=content,
additional_kwargs=additional_kwargs,
tool_calls=tool_calls,
invalid_tool_calls=invalid_tool_calls,
)
def _convert_message_to_dict(message: BaseMessage) -> dict:
"""Convert a LangChain message to a dictionary.
Args:
message: The LangChain message.
Returns:
The dictionary.
"""
message_dict: Dict[str, Any]
if isinstance(message, ChatMessage):
message_dict = {"role": message.role, "content": message.content}
elif isinstance(message, HumanMessage):
message_dict = {"role": "user", "content": message.content}
elif isinstance(message, AIMessage):
message_dict = {"role": "assistant", "content": message.content}
if "function_call" in message.additional_kwargs:
message_dict["function_call"] = message.additional_kwargs["function_call"]
# If function call only, content is None not empty string
if message_dict["content"] == "":
message_dict["content"] = None
if "tool_calls" in message.additional_kwargs:
message_dict["tool_calls"] = message.additional_kwargs["tool_calls"]
# If tool calls only, content is None not empty string
if message_dict["content"] == "":
message_dict["content"] = None
elif isinstance(message, SystemMessage):
message_dict = {"role": "system", "content": message.content}
elif isinstance(message, FunctionMessage):
message_dict = {
"role": "function",
"content": message.content,
"name": message.name,
}
elif isinstance(message, ToolMessage):
message_dict = {
"role": "tool",
"content": message.content,
"tool_call_id": "None",
}
else:
raise TypeError(f"Got unknown type {message}")
if "name" in message.additional_kwargs:
message_dict["name"] = message.additional_kwargs["name"]
return message_dict
def _convert_delta_to_message_chunk(
_dict: Mapping[str, Any], default_class: Type[BaseMessageChunk]
) -> BaseMessageChunk:
role = cast(str, _dict.get("role"))
content = cast(str, _dict.get("content") or "")
additional_kwargs: Dict = {}
tool_call_chunks: List[ToolCallChunk] = []
if _dict.get("function_call"):
function_call = dict(_dict["function_call"])
if "name" in function_call and function_call["name"] is None:
function_call["name"] = ""
additional_kwargs["function_call"] = function_call
if raw_tool_calls := _dict.get("tool_calls"):
additional_kwargs["tool_calls"] = raw_tool_calls
for rtc in raw_tool_calls:
try:
tool_call_chunks.append(
create_tool_call_chunk(
name=rtc["function"].get("name"),
args=rtc["function"].get("arguments"),
id=rtc.get("id"),
index=rtc.get("index"),
)
)
except KeyError:
pass
if role == "user" or default_class == HumanMessageChunk:
return HumanMessageChunk(content=content)
elif role == "assistant" or default_class == AIMessageChunk:
return AIMessageChunk(
content=content,
additional_kwargs=additional_kwargs,
tool_call_chunks=tool_call_chunks, # type: ignore[arg-type]
)
elif role == "system" or default_class == SystemMessageChunk:
return SystemMessageChunk(content=content)
elif role == "function" or default_class == FunctionMessageChunk:
return FunctionMessageChunk(content=content, name=_dict["name"])
elif role == "tool" or default_class == ToolMessageChunk:
return ToolMessageChunk(content=content, tool_call_id=_dict["tool_call_id"])
elif role or default_class == ChatMessageChunk:
return ChatMessageChunk(content=content, role=role)
else:
return default_class(content=content) # type: ignore
class _FunctionCall(TypedDict):
name: str
class ChatWatsonx(BaseChatModel):
"""
IBM watsonx.ai large language chat models.
To use, you should have ``langchain_ibm`` python package installed,
and the environment variable ``WATSONX_APIKEY`` set with your API key, or pass
it as a named parameter to the constructor.
Example:
.. code-block:: python
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 100,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.TEMPERATURE: 0.5,
GenTextParamsMetaNames.TOP_K: 50,
GenTextParamsMetaNames.TOP_P: 1,
}
from langchain_ibm import ChatWatsonx
watsonx_llm = ChatWatsonx(
model_id="meta-llama/llama-3-70b-instruct",
url="https://us-south.ml.cloud.ibm.com",
apikey="*****",
project_id="*****",
params=parameters,
)
"""
model_id: str = ""
"""Type of model to use."""
deployment_id: str = ""
"""Type of deployed model to use."""
project_id: str = ""
"""ID of the Watson Studio project."""
space_id: str = ""
"""ID of the Watson Studio space."""
url: Optional[SecretStr] = None
"""Url to Watson Machine Learning or CPD instance"""
apikey: Optional[SecretStr] = None
"""Apikey to Watson Machine Learning or CPD instance"""
token: Optional[SecretStr] = None
"""Token to CPD instance"""
password: Optional[SecretStr] = None
"""Password to CPD instance"""
username: Optional[SecretStr] = None
"""Username to CPD instance"""
instance_id: Optional[SecretStr] = None
"""Instance_id of CPD instance"""
version: Optional[SecretStr] = None
"""Version of CPD instance"""
params: Optional[dict] = None
"""Chat Model parameters to use during generate requests."""
verify: Union[str, bool] = ""
"""User can pass as verify one of following:
the path to a CA_BUNDLE file
the path of directory with certificates of trusted CAs
True - default path to truststore will be taken
False - no verification will be made"""
streaming: bool = False
""" Whether to stream the results or not. """
watsonx_model: ModelInference = Field(default=None, exclude=True) #: :meta private:
class Config:
"""Configuration for this pydantic object."""
allow_population_by_field_name = True
@classmethod
def is_lc_serializable(cls) -> bool:
return False
@property
def _llm_type(self) -> str:
"""Return type of chat model."""
return "watsonx-chat"
def _get_ls_params(
self, stop: Optional[List[str]] = None, **kwargs: Any
) -> LangSmithParams:
"""Get standard params for tracing."""
params = super()._get_ls_params(stop=stop, **kwargs)
params["ls_provider"] = "together"
params["ls_model_name"] = self.model_id
return params
@property
def lc_secrets(self) -> Dict[str, str]:
"""A map of constructor argument names to secret ids.
For example:
{
"url": "WATSONX_URL",
"apikey": "WATSONX_APIKEY",
"token": "WATSONX_TOKEN",
"password": "WATSONX_PASSWORD",
"username": "WATSONX_USERNAME",
"instance_id": "WATSONX_INSTANCE_ID",
}
"""
return {
"url": "WATSONX_URL",
"apikey": "WATSONX_APIKEY",
"token": "WATSONX_TOKEN",
"password": "WATSONX_PASSWORD",
"username": "WATSONX_USERNAME",
"instance_id": "WATSONX_INSTANCE_ID",
}
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""Validate that credentials and python package exists in environment."""
values["url"] = convert_to_secret_str(
get_from_dict_or_env(values, "url", "WATSONX_URL")
)
if "cloud.ibm.com" in values.get("url", "").get_secret_value():
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
else:
if (
not values["token"]
and "WATSONX_TOKEN" not in os.environ
and not values["password"]
and "WATSONX_PASSWORD" not in os.environ
and not values["apikey"]
and "WATSONX_APIKEY" not in os.environ
):
raise ValueError(
"Did not find 'token', 'password' or 'apikey',"
" please add an environment variable"
" `WATSONX_TOKEN`, 'WATSONX_PASSWORD' or 'WATSONX_APIKEY' "
"which contains it,"
" or pass 'token', 'password' or 'apikey'"
" as a named parameter."
)
elif values["token"] or "WATSONX_TOKEN" in os.environ:
values["token"] = convert_to_secret_str(
get_from_dict_or_env(values, "token", "WATSONX_TOKEN")
)
elif values["password"] or "WATSONX_PASSWORD" in os.environ:
values["password"] = convert_to_secret_str(
get_from_dict_or_env(values, "password", "WATSONX_PASSWORD")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
elif values["apikey"] or "WATSONX_APIKEY" in os.environ:
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
if not values["instance_id"] or "WATSONX_INSTANCE_ID" not in os.environ:
values["instance_id"] = convert_to_secret_str(
get_from_dict_or_env(values, "instance_id", "WATSONX_INSTANCE_ID")
)
credentials = Credentials(
url=values["url"].get_secret_value() if values["url"] else None,
api_key=values["apikey"].get_secret_value() if values["apikey"] else None,
token=values["token"].get_secret_value() if values["token"] else None,
password=values["password"].get_secret_value()
if values["password"]
else None,
username=values["username"].get_secret_value()
if values["username"]
else None,
instance_id=values["instance_id"].get_secret_value()
if values["instance_id"]
else None,
version=values["version"].get_secret_value() if values["version"] else None,
verify=values["verify"],
)
watsonx_chat = ModelInference(
model_id=values["model_id"],
deployment_id=values["deployment_id"],
credentials=credentials,
params=values["params"],
project_id=values["project_id"],
space_id=values["space_id"],
)
values["watsonx_model"] = watsonx_chat
return values
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> ChatResult:
should_stream = stream if stream is not None else self.streaming
if should_stream:
stream_iter = self._stream(
messages, stop=stop, run_manager=run_manager, **kwargs
)
return generate_from_stream(stream_iter)
message_dicts, params = self._create_message_dicts(messages, stop, **kwargs)
chat_prompt = self._create_chat_prompt(message_dicts)
tools = kwargs.get("tools")
if tools:
chat_prompt = f"""[AVAILABLE_TOOLS]
{json.dumps(tools[0], indent=2)}
[/AVAILABLE_TOOLS]
[INST]<<SYS>>You are Mixtral Chat function calling, an AI language model developed by
Mistral AI. You are a cautious assistant. You carefully follow instructions. You are
helpful and harmless and you follow ethical guidelines and promote positive behavior.
<</SYS>>
To use these tools you must always respond in JSON format containing `"type"` and
`"function"` key-value pairs. Also `"function"` key-value pair always containing
`"name"` and `"arguments"` key-value pairs.
Between subsequent JSONs should be one blank line.
Remember, even when answering to the user, you must still use this only JSON format!
{chat_prompt}[/INST]"""
if "tools" in kwargs:
del kwargs["tools"]
if "tool_choice" in kwargs:
del kwargs["tool_choice"]
response = self.watsonx_model.generate(
prompt=chat_prompt, **(kwargs | {"params": params})
)
return self._create_chat_result(response)
def _stream(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> Iterator[ChatGenerationChunk]:
message_dicts, params = self._create_message_dicts(messages, stop, **kwargs)
chat_prompt = self._create_chat_prompt(message_dicts)
tools = kwargs.get("tools")
if tools:
chat_prompt = f"""[AVAILABLE_TOOLS]
{json.dumps(tools[0], indent=2)}
[/AVAILABLE_TOOLS]
[INST]<<SYS>>You are Mixtral Chat function calling, an AI language model developed by
Mistral AI. You are a cautious assistant. You carefully follow instructions. You are
helpful and harmless and you follow ethical guidelines and promote positive behavior.
<</SYS>>
To use these tools you must always respond in JSON format containing `"type"` and
`"function"` key-value pairs. Also `"function"` key-value pair always containing
`"name"` and `"arguments"` key-value pairs.
Between subsequent JSONs should be one blank line.
Remember, even when answering to the user, you must still use this only JSON format!
{chat_prompt}[/INST]"""
if "tools" in kwargs:
del kwargs["tools"]
if "tool_choice" in kwargs:
del kwargs["tool_choice"]
for chunk in self.watsonx_model.generate_text_stream(
prompt=chat_prompt, raw_response=True, **(kwargs | {"params": params})
):
if not isinstance(chunk, dict):
chunk = chunk.dict()
if len(chunk["results"]) == 0:
continue
choice = chunk["results"][0]
chunk = AIMessageChunk(
content=choice["generated_text"],
)
generation_info = {}
if finish_reason := choice.get("stop_reason"):
generation_info["finish_reason"] = finish_reason
logprobs = choice.get("logprobs")
if logprobs:
generation_info["logprobs"] = logprobs
chunk = ChatGenerationChunk(
message=chunk, generation_info=generation_info or None
)
if run_manager:
run_manager.on_llm_new_token(
chunk.content, chunk=chunk, logprobs=logprobs
)
yield chunk
def _create_chat_prompt(self, messages: List[Dict[str, Any]]) -> str:
prompt = ""
if self.model_id in ["ibm/granite-13b-chat-v1", "ibm/granite-13b-chat-v2"]:
for message in messages:
if message["role"] == "system":
prompt += "<|system|>\n" + message["content"] + "\n\n"
elif message["role"] == "assistant":
prompt += "<|assistant|>\n" + message["content"] + "\n\n"
elif message["role"] == "function":
prompt += "<|function|>\n" + message["content"] + "\n\n"
elif message["role"] == "tool":
prompt += "<|tool|>\n" + message["content"] + "\n\n"
else:
prompt += "<|user|>:\n" + message["content"] + "\n\n"
prompt += "<|assistant|>\n"
elif self.model_id in [
"meta-llama/llama-2-13b-chat",
"meta-llama/llama-2-70b-chat",
]:
for message in messages:
if message["role"] == "system":
prompt += "[INST] <<SYS>>\n" + message["content"] + "<</SYS>>\n\n"
elif message["role"] == "assistant":
prompt += message["content"] + "\n[INST]\n\n"
else:
prompt += message["content"] + "\n[/INST]\n"
else:
prompt = ChatPromptValue(
messages=convert_to_messages(messages) + [AIMessage(content="")]
).to_string()
return prompt
def _create_message_dicts(
self, messages: List[BaseMessage], stop: Optional[List[str]], **kwargs: Any
) -> Tuple[List[Dict[str, Any]], Dict[str, Any]]:
params = {**self.params} if self.params else {}
params = params | {**kwargs.get("params", {})}
if stop is not None:
if params and "stop_sequences" in params:
raise ValueError(
"`stop_sequences` found in both the input and default params."
)
params = (params or {}) | {"stop_sequences": stop}
message_dicts = [_convert_message_to_dict(m) for m in messages]
return message_dicts, params
def _create_chat_result(self, response: Union[dict]) -> ChatResult:
generations = []
sum_of_total_generated_tokens = 0
sum_of_total_input_tokens = 0
if response.get("error"):
raise ValueError(response.get("error"))
for res in response["results"]:
message = _convert_dict_to_message(res)
generation_info = dict(finish_reason=res.get("stop_reason"))
if "logprobs" in res:
generation_info["logprobs"] = res["logprobs"]
if "generated_token_count" in res:
sum_of_total_generated_tokens += res["generated_token_count"]
if "input_token_count" in res:
sum_of_total_input_tokens += res["input_token_count"]
total_token = sum_of_total_generated_tokens + sum_of_total_input_tokens
if total_token and isinstance(message, AIMessage):
message.usage_metadata = {
"input_tokens": sum_of_total_input_tokens,
"output_tokens": sum_of_total_generated_tokens,
"total_tokens": total_token,
}
gen = ChatGeneration(
message=message,
generation_info=generation_info,
)
generations.append(gen)
token_usage = {
"generated_token_count": sum_of_total_generated_tokens,
"input_token_count": sum_of_total_input_tokens,
}
llm_output = {
"token_usage": token_usage,
"model_name": self.model_id,
"system_fingerprint": response.get("system_fingerprint", ""),
}
return ChatResult(generations=generations, llm_output=llm_output)
def bind_functions(
self,
functions: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
function_call: Optional[
Union[_FunctionCall, str, Literal["auto", "none"]]
] = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
"""Bind functions (and other objects) to this chat model.
Assumes model is compatible with IBM watsonx.ai function-calling API.
Args:
functions: A list of function 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.
function_call: Which function to require the model to call.
Must be the name of the single provided function or
"auto" to automatically determine which function to call
(if any).
**kwargs: Any additional parameters to pass to the
:class:`~langchain.runnable.Runnable` constructor.
"""
formatted_functions = [convert_to_openai_function(fn) for fn in functions]
if function_call is not None:
function_call = (
{"name": function_call}
if isinstance(function_call, str)
and function_call not in ("auto", "none")
else function_call
)
if isinstance(function_call, dict) and len(formatted_functions) != 1:
raise ValueError(
"When specifying `function_call`, you must provide exactly one "
"function."
)
if (
isinstance(function_call, dict)
and formatted_functions[0]["name"] != function_call["name"]
):
raise ValueError(
f"Function call {function_call} was specified, but the only "
f"provided function was {formatted_functions[0]['name']}."
)
kwargs = {**kwargs, "function_call": function_call}
return super().bind(
functions=formatted_functions,
**kwargs,
)
def bind_tools(
self,
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
"""Bind tool-like objects to this chat model.
Args:
tools: A list of tool definitions to bind to this chat model.
Can be a dictionary, pydantic model, callable, or BaseTool. Pydantic
models, callables, and BaseTools will be automatically converted to
their schema dictionary representation.
**kwargs: Any additional parameters to pass to the
:class:`~langchain.runnable.Runnable` constructor.
"""
bind_tools_supported_models = ["mistralai/mixtral-8x7b-instruct-v01"]
if self.model_id not in bind_tools_supported_models:
raise Warning(
f"bind_tools() method for ChatWatsonx support only "
f"following models: {bind_tools_supported_models}"
)
formatted_tools = [convert_to_openai_tool(tool) for tool in tools]
return super().bind(tools=formatted_tools, **kwargs)
def with_structured_output(
self,
schema: Optional[Union[Dict, Type[BaseModel]]] = None,
*,
method: Literal["function_calling", "json_mode"] = "function_calling",
include_raw: bool = False,
**kwargs: Any,
) -> Runnable[LanguageModelInput, Union[Dict, BaseModel]]:
"""Model wrapper that returns outputs formatted to match the given schema.
Args:
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 IBM watsonx.ai function-calling spec.
method: The method for steering model generation, either "function_calling"
or "json_mode". If "function_calling" then the schema will be converted
to an IBM watsonx.ai function and the returned model will make use of the
function-calling API. If "json_mode" then IBM watsonx.ai'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.
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:
A Runnable that takes any ChatModel input and returns as output:
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.
Example: Function-calling, Pydantic schema (method="function_calling", include_raw=False):
.. code-block:: python
from langchain_ibm import ChatWatsonx
from langchain_core.pydantic_v1 import BaseModel
class AnswerWithJustification(BaseModel):
'''An answer to the user question along with justification for the answer.'''
answer: str
justification: str
llm = ChatWatsonx(...)
structured_llm = llm.with_structured_output(AnswerWithJustification)
structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> AnswerWithJustification(
# answer='They weigh the same',
# justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'
# )
Example: Function-calling, Pydantic schema (method="function_calling", include_raw=True):
.. code-block:: python
from langchain_ibm import ChatWatsonx
from langchain_core.pydantic_v1 import BaseModel
class AnswerWithJustification(BaseModel):
'''An answer to the user question along with justification for the answer.'''
answer: str
justification: str
llm = ChatWatsonx(...)
structured_llm = llm.with_structured_output(AnswerWithJustification, include_raw=True)
structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> {
# 'raw': AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_Ao02pnFYXD6GN1yzc0uXPsvF', 'function': {'arguments': '{"answer":"They weigh the same.","justification":"Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ."}', 'name': 'AnswerWithJustification'}, 'type': 'function'}]}),
# 'parsed': AnswerWithJustification(answer='They weigh the same.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume or density of the objects may differ.'),
# 'parsing_error': None
# }
Example: Function-calling, dict schema (method="function_calling", include_raw=False):
.. code-block:: python
from langchain_ibm import ChatWatsonx
from langchain_core.pydantic_v1 import BaseModel
from langchain_core.utils.function_calling import convert_to_openai_tool
class AnswerWithJustification(BaseModel):
'''An answer to the user question along with justification for the answer.'''
answer: str
justification: str
dict_schema = convert_to_openai_tool(AnswerWithJustification)
llm = ChatWatsonx(...)
structured_llm = llm.with_structured_output(dict_schema)
structured_llm.invoke("What weighs more a pound of bricks or a pound of feathers")
# -> {
# 'answer': 'They weigh the same',
# 'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The weight is the same, but the volume and density of the two substances differ.'
# }
Example: JSON mode, Pydantic schema (method="json_mode", include_raw=True):
.. code-block::
from langchain_ibm import ChatWatsonx
from langchain_core.pydantic_v1 import BaseModel
class AnswerWithJustification(BaseModel):
answer: str
justification: str
llm = ChatWatsonx(...)
structured_llm = llm.with_structured_output(
AnswerWithJustification,
method="json_mode",
include_raw=True
)
structured_llm.invoke(
"Answer the following question. "
"Make sure to return a JSON blob with keys 'answer' and 'justification'.\n\n"
"What's heavier a pound of bricks or a pound of feathers?"
)
# -> {
# 'raw': AIMessage(content='{\n "answer": "They are both the same weight.",\n "justification": "Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight." \n}'),
# 'parsed': AnswerWithJustification(answer='They are both the same weight.', justification='Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight.'),
# 'parsing_error': None
# }
Example: JSON mode, no schema (schema=None, method="json_mode", include_raw=True):
.. code-block::
from langchain_ibm import ChatWatsonx
structured_llm = llm.with_structured_output(method="json_mode", include_raw=True)
structured_llm.invoke(
"Answer the following question. "
"Make sure to return a JSON blob with keys 'answer' and 'justification'.\n\n"
"What's heavier a pound of bricks or a pound of feathers?"
)
# -> {
# 'raw': AIMessage(content='{\n "answer": "They are both the same weight.",\n "justification": "Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight." \n}'),
# 'parsed': {
# 'answer': 'They are both the same weight.',
# 'justification': 'Both a pound of bricks and a pound of feathers weigh one pound. The difference lies in the volume and density of the materials, not the weight.'
# },
# 'parsing_error': None
# }
""" # noqa: E501
if kwargs:
raise ValueError(f"Received unsupported arguments {kwargs}")
is_pydantic_schema = _is_pydantic_class(schema)
if method == "function_calling":
if schema is None:
raise ValueError(
"schema must be specified when method is 'function_calling'. "
"Received None."
)
llm = self.bind_tools([schema], tool_choice=True)
if is_pydantic_schema:
output_parser: OutputParserLike = PydanticToolsParser(
tools=[schema], # type: ignore[list-item]
first_tool_only=True, # type: ignore[list-item]
)
else:
key_name = convert_to_openai_tool(schema)["function"]["name"]
output_parser = JsonOutputKeyToolsParser(
key_name=key_name, first_tool_only=True
)
elif method == "json_mode":
llm = self.bind(response_format={"type": "json_object"})
output_parser = (
PydanticOutputParser(pydantic_object=schema) # type: ignore[type-var, arg-type]
if is_pydantic_schema
else JsonOutputParser()
)
else:
raise ValueError(
f"Unrecognized method argument. Expected one of 'function_calling' or "
f"'json_format'. Received: '{method}'"
)
if include_raw:
parser_assign = RunnablePassthrough.assign(
parsed=itemgetter("raw") | output_parser, parsing_error=lambda _: None
)
parser_none = RunnablePassthrough.assign(parsed=lambda _: None)
parser_with_fallback = parser_assign.with_fallbacks(
[parser_none], exception_key="parsing_error"
)
return RunnableMap(raw=llm) | parser_with_fallback
else:
return llm | output_parser
def _is_pydantic_class(obj: Any) -> bool:
return isinstance(obj, type) and issubclass(obj, BaseModel)

173
libs/partners/ibm/langchain_ibm/embeddings.py
View File

@ -1,173 +0,0 @@
import os
from typing import Dict, List, Optional, Union
from ibm_watsonx_ai import APIClient, Credentials # type: ignore
from ibm_watsonx_ai.foundation_models.embeddings import Embeddings # type: ignore
from langchain_core.embeddings import Embeddings as LangChainEmbeddings
from langchain_core.pydantic_v1 import (
BaseModel,
Extra,
Field,
SecretStr,
root_validator,
)
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env
class WatsonxEmbeddings(BaseModel, LangChainEmbeddings):
"""IBM WatsonX.ai embedding models."""
model_id: str = ""
"""Type of model to use."""
project_id: str = ""
"""ID of the Watson Studio project."""
space_id: str = ""
"""ID of the Watson Studio space."""
url: Optional[SecretStr] = None
"""Url to Watson Machine Learning or CPD instance"""
apikey: Optional[SecretStr] = None
"""Apikey to Watson Machine Learning or CPD instance"""
token: Optional[SecretStr] = None
"""Token to CPD instance"""
password: Optional[SecretStr] = None
"""Password to CPD instance"""
username: Optional[SecretStr] = None
"""Username to CPD instance"""
instance_id: Optional[SecretStr] = None
"""Instance_id of CPD instance"""
version: Optional[SecretStr] = None
"""Version of CPD instance"""
params: Optional[dict] = None
"""Model parameters to use during generate requests."""
verify: Union[str, bool, None] = None
"""User can pass as verify one of following:
the path to a CA_BUNDLE file
the path of directory with certificates of trusted CAs
True - default path to truststore will be taken
False - no verification will be made"""
watsonx_embed: Embeddings = Field(default=None) #: :meta private:
watsonx_client: APIClient = Field(default=None) #: :meta private:
class Config:
"""Configuration for this pydantic object."""
extra = Extra.forbid
arbitrary_types_allowed = True
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""Validate that credentials and python package exists in environment."""
if isinstance(values.get("watsonx_client"), APIClient):
watsonx_embed = Embeddings(
model_id=values["model_id"],
params=values["params"],
api_client=values["watsonx_client"],
project_id=values["project_id"],
space_id=values["space_id"],
verify=values["verify"],
)
values["watsonx_embed"] = watsonx_embed
else:
values["url"] = convert_to_secret_str(
get_from_dict_or_env(values, "url", "WATSONX_URL")
)
if "cloud.ibm.com" in values.get("url", "").get_secret_value():
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
else:
if (
not values["token"]
and "WATSONX_TOKEN" not in os.environ
and not values["password"]
and "WATSONX_PASSWORD" not in os.environ
and not values["apikey"]
and "WATSONX_APIKEY" not in os.environ
):
raise ValueError(
"Did not find 'token', 'password' or 'apikey',"
" please add an environment variable"
" `WATSONX_TOKEN`, 'WATSONX_PASSWORD' or 'WATSONX_APIKEY' "
"which contains it,"
" or pass 'token', 'password' or 'apikey'"
" as a named parameter."
)
elif values["token"] or "WATSONX_TOKEN" in os.environ:
values["token"] = convert_to_secret_str(
get_from_dict_or_env(values, "token", "WATSONX_TOKEN")
)
elif values["password"] or "WATSONX_PASSWORD" in os.environ:
values["password"] = convert_to_secret_str(
get_from_dict_or_env(values, "password", "WATSONX_PASSWORD")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
elif values["apikey"] or "WATSONX_APIKEY" in os.environ:
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
if not values["instance_id"] or "WATSONX_INSTANCE_ID" not in os.environ:
values["instance_id"] = convert_to_secret_str(
get_from_dict_or_env(
values, "instance_id", "WATSONX_INSTANCE_ID"
)
)
credentials = Credentials(
url=values["url"].get_secret_value() if values["url"] else None,
api_key=values["apikey"].get_secret_value()
if values["apikey"]
else None,
token=values["token"].get_secret_value() if values["token"] else None,
password=values["password"].get_secret_value()
if values["password"]
else None,
username=values["username"].get_secret_value()
if values["username"]
else None,
instance_id=values["instance_id"].get_secret_value()
if values["instance_id"]
else None,
version=values["version"].get_secret_value()
if values["version"]
else None,
verify=values["verify"],
)
watsonx_embed = Embeddings(
model_id=values["model_id"],
params=values["params"],
credentials=credentials,
project_id=values["project_id"],
space_id=values["space_id"],
)
values["watsonx_embed"] = watsonx_embed
return values
def embed_documents(self, texts: List[str]) -> List[List[float]]:
"""Embed search docs."""
return self.watsonx_embed.embed_documents(texts=texts)
def embed_query(self, text: str) -> List[float]:
"""Embed query text."""
return self.embed_documents([text])[0]

428
libs/partners/ibm/langchain_ibm/llms.py
View File

@ -1,428 +0,0 @@
import logging
import os
from typing import Any, Dict, Iterator, List, Mapping, Optional, Union
from ibm_watsonx_ai import Credentials # type: ignore
from ibm_watsonx_ai.foundation_models import Model, ModelInference # type: ignore
from langchain_core.callbacks import CallbackManagerForLLMRun
from langchain_core.language_models.llms import BaseLLM
from langchain_core.outputs import Generation, GenerationChunk, LLMResult
from langchain_core.pydantic_v1 import Extra, Field, SecretStr, root_validator
from langchain_core.utils import convert_to_secret_str, get_from_dict_or_env
logger = logging.getLogger(__name__)
class WatsonxLLM(BaseLLM):
"""
IBM WatsonX.ai large language models.
To use, you should have ``langchain_ibm`` python package installed,
and the environment variable ``WATSONX_APIKEY`` set with your API key, or pass
it as a named parameter to the constructor.
Example:
.. code-block:: python
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 100,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.TEMPERATURE: 0.5,
GenTextParamsMetaNames.TOP_K: 50,
GenTextParamsMetaNames.TOP_P: 1,
}
from langchain_ibm import WatsonxLLM
watsonx_llm = WatsonxLLM(
model_id="google/flan-ul2",
url="https://us-south.ml.cloud.ibm.com",
apikey="*****",
project_id="*****",
params=parameters,
)
"""
model_id: str = ""
"""Type of model to use."""
deployment_id: str = ""
"""Type of deployed model to use."""
project_id: str = ""
"""ID of the Watson Studio project."""
space_id: str = ""
"""ID of the Watson Studio space."""
url: Optional[SecretStr] = None
"""Url to Watson Machine Learning or CPD instance"""
apikey: Optional[SecretStr] = None
"""Apikey to Watson Machine Learning or CPD instance"""
token: Optional[SecretStr] = None
"""Token to CPD instance"""
password: Optional[SecretStr] = None
"""Password to CPD instance"""
username: Optional[SecretStr] = None
"""Username to CPD instance"""
instance_id: Optional[SecretStr] = None
"""Instance_id of CPD instance"""
version: Optional[SecretStr] = None
"""Version of CPD instance"""
params: Optional[dict] = None
"""Model parameters to use during generate requests."""
verify: Union[str, bool, None] = None
"""User can pass as verify one of following:
the path to a CA_BUNDLE file
the path of directory with certificates of trusted CAs
True - default path to truststore will be taken
False - no verification will be made"""
streaming: bool = False
""" Whether to stream the results or not. """
watsonx_model: ModelInference = Field(default=None, exclude=True) #: :meta private:
class Config:
"""Configuration for this pydantic object."""
extra = Extra.forbid
@classmethod
def is_lc_serializable(cls) -> bool:
return False
@property
def lc_secrets(self) -> Dict[str, str]:
"""A map of constructor argument names to secret ids.
For example:
{
"url": "WATSONX_URL",
"apikey": "WATSONX_APIKEY",
"token": "WATSONX_TOKEN",
"password": "WATSONX_PASSWORD",
"username": "WATSONX_USERNAME",
"instance_id": "WATSONX_INSTANCE_ID",
}
"""
return {
"url": "WATSONX_URL",
"apikey": "WATSONX_APIKEY",
"token": "WATSONX_TOKEN",
"password": "WATSONX_PASSWORD",
"username": "WATSONX_USERNAME",
"instance_id": "WATSONX_INSTANCE_ID",
}
@root_validator()
def validate_environment(cls, values: Dict) -> Dict:
"""Validate that credentials and python package exists in environment."""
if isinstance(values.get("watsonx_model"), (ModelInference, Model)):
values["model_id"] = getattr(values["watsonx_model"], "model_id")
values["deployment_id"] = getattr(
values["watsonx_model"], "deployment_id", ""
)
values["project_id"] = getattr(
getattr(values["watsonx_model"], "_client"),
"default_project_id",
)
values["space_id"] = getattr(
getattr(values["watsonx_model"], "_client"), "default_space_id"
)
values["params"] = getattr(values["watsonx_model"], "params")
else:
values["url"] = convert_to_secret_str(
get_from_dict_or_env(values, "url", "WATSONX_URL")
)
if "cloud.ibm.com" in values.get("url", "").get_secret_value():
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
else:
if (
not values["token"]
and "WATSONX_TOKEN" not in os.environ
and not values["password"]
and "WATSONX_PASSWORD" not in os.environ
and not values["apikey"]
and "WATSONX_APIKEY" not in os.environ
):
raise ValueError(
"Did not find 'token', 'password' or 'apikey',"
" please add an environment variable"
" `WATSONX_TOKEN`, 'WATSONX_PASSWORD' or 'WATSONX_APIKEY' "
"which contains it,"
" or pass 'token', 'password' or 'apikey'"
" as a named parameter."
)
elif values["token"] or "WATSONX_TOKEN" in os.environ:
values["token"] = convert_to_secret_str(
get_from_dict_or_env(values, "token", "WATSONX_TOKEN")
)
elif values["password"] or "WATSONX_PASSWORD" in os.environ:
values["password"] = convert_to_secret_str(
get_from_dict_or_env(values, "password", "WATSONX_PASSWORD")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
elif values["apikey"] or "WATSONX_APIKEY" in os.environ:
values["apikey"] = convert_to_secret_str(
get_from_dict_or_env(values, "apikey", "WATSONX_APIKEY")
)
values["username"] = convert_to_secret_str(
get_from_dict_or_env(values, "username", "WATSONX_USERNAME")
)
if not values["instance_id"] or "WATSONX_INSTANCE_ID" not in os.environ:
values["instance_id"] = convert_to_secret_str(
get_from_dict_or_env(
values, "instance_id", "WATSONX_INSTANCE_ID"
)
)
credentials = Credentials(
url=values["url"].get_secret_value() if values["url"] else None,
api_key=values["apikey"].get_secret_value()
if values["apikey"]
else None,
token=values["token"].get_secret_value() if values["token"] else None,
password=values["password"].get_secret_value()
if values["password"]
else None,
username=values["username"].get_secret_value()
if values["username"]
else None,
instance_id=values["instance_id"].get_secret_value()
if values["instance_id"]
else None,
version=values["version"].get_secret_value()
if values["version"]
else None,
verify=values["verify"],
)
watsonx_model = ModelInference(
model_id=values["model_id"],
deployment_id=values["deployment_id"],
credentials=credentials,
params=values["params"],
project_id=values["project_id"],
space_id=values["space_id"],
)
values["watsonx_model"] = watsonx_model
return values
@property
def _identifying_params(self) -> Mapping[str, Any]:
"""Get the identifying parameters."""
return {
"model_id": self.model_id,
"deployment_id": self.deployment_id,
"params": self.params,
"project_id": self.project_id,
"space_id": self.space_id,
}
@property
def _llm_type(self) -> str:
"""Return type of llm."""
return "IBM watsonx.ai"
@staticmethod
def _extract_token_usage(
response: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
if response is None:
return {"generated_token_count": 0, "input_token_count": 0}
input_token_count = 0
generated_token_count = 0
def get_count_value(key: str, result: Dict[str, Any]) -> int:
return result.get(key, 0) or 0
for res in response:
results = res.get("results")
if results:
input_token_count += get_count_value("input_token_count", results[0])
generated_token_count += get_count_value(
"generated_token_count", results[0]
)
return {
"generated_token_count": generated_token_count,
"input_token_count": input_token_count,
}
def _get_chat_params(
self, stop: Optional[List[str]] = None, **kwargs: Any
) -> Optional[Dict[str, Any]]:
params = {**self.params} if self.params else {}
params = params | {**kwargs.get("params", {})}
if stop is not None:
if params and "stop_sequences" in params:
raise ValueError(
"`stop_sequences` found in both the input and default params."
)
params = (params or {}) | {"stop_sequences": stop}
return params
def _create_llm_result(self, response: List[dict]) -> LLMResult:
"""Create the LLMResult from the choices and prompts."""
generations = []
for res in response:
results = res.get("results")
if results:
finish_reason = results[0].get("stop_reason")
gen = Generation(
text=results[0].get("generated_text"),
generation_info={"finish_reason": finish_reason},
)
generations.append([gen])
final_token_usage = self._extract_token_usage(response)
llm_output = {
"token_usage": final_token_usage,
"model_id": self.model_id,
"deployment_id": self.deployment_id,
}
return LLMResult(generations=generations, llm_output=llm_output)
def _stream_response_to_generation_chunk(
self,
stream_response: Dict[str, Any],
) -> GenerationChunk:
"""Convert a stream response to a generation chunk."""
if not stream_response["results"]:
return GenerationChunk(text="")
return GenerationChunk(
text=stream_response["results"][0]["generated_text"],
generation_info=dict(
finish_reason=stream_response["results"][0].get("stop_reason", None),
llm_output={
"model_id": self.model_id,
"deployment_id": self.deployment_id,
},
),
)
def _call(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> str:
"""Call the IBM watsonx.ai inference endpoint.
Args:
prompt: The prompt to pass into the model.
stop: Optional list of stop words to use when generating.
run_manager: Optional callback manager.
Returns:
The string generated by the model.
Example:
.. code-block:: python
response = watsonx_llm.invoke("What is a molecule")
"""
result = self._generate(
prompts=[prompt], stop=stop, run_manager=run_manager, **kwargs
)
return result.generations[0][0].text
def _generate(
self,
prompts: List[str],
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
stream: Optional[bool] = None,
**kwargs: Any,
) -> LLMResult:
"""Call the IBM watsonx.ai inference endpoint which then generate the response.
Args:
prompts: List of strings (prompts) to pass into the model.
stop: Optional list of stop words to use when generating.
run_manager: Optional callback manager.
Returns:
The full LLMResult output.
Example:
.. code-block:: python
response = watsonx_llm.generate(["What is a molecule"])
"""
params = self._get_chat_params(stop=stop, **kwargs)
should_stream = stream if stream is not None else self.streaming
if should_stream:
if len(prompts) > 1:
raise ValueError(
f"WatsonxLLM currently only supports single prompt, got {prompts}"
)
generation = GenerationChunk(text="")
stream_iter = self._stream(
prompts[0], stop=stop, run_manager=run_manager, **kwargs
)
for chunk in stream_iter:
if generation is None:
generation = chunk
else:
generation += chunk
assert generation is not None
if isinstance(generation.generation_info, dict):
llm_output = generation.generation_info.pop("llm_output")
return LLMResult(generations=[[generation]], llm_output=llm_output)
return LLMResult(generations=[[generation]])
else:
response = self.watsonx_model.generate(
prompt=prompts, **(kwargs | {"params": params})
)
return self._create_llm_result(response)
def _stream(
self,
prompt: str,
stop: Optional[List[str]] = None,
run_manager: Optional[CallbackManagerForLLMRun] = None,
**kwargs: Any,
) -> Iterator[GenerationChunk]:
"""Call the IBM watsonx.ai inference endpoint which then streams the response.
Args:
prompt: The prompt to pass into the model.
stop: Optional list of stop words to use when generating.
run_manager: Optional callback manager.
Returns:
The iterator which yields generation chunks.
Example:
.. code-block:: python
response = watsonx_llm.stream("What is a molecule")
for chunk in response:
print(chunk, end='')
"""
params = self._get_chat_params(stop=stop, **kwargs)
for stream_resp in self.watsonx_model.generate_text_stream(
prompt=prompt, raw_response=True, **(kwargs | {"params": params})
):
if not isinstance(stream_resp, dict):
stream_resp = stream_resp.dict()
chunk = self._stream_response_to_generation_chunk(stream_resp)
if run_manager:
run_manager.on_llm_new_token(chunk.text, chunk=chunk)
yield chunk
def get_num_tokens(self, text: str) -> int:
response = self.watsonx_model.tokenize(text, return_tokens=False)
return response["result"]["token_count"]
def get_token_ids(self, text: str) -> List[int]:
raise NotImplementedError("API does not support returning token ids.")

0
libs/partners/ibm/langchain_ibm/py.typed
View File

1150
libs/partners/ibm/poetry.lock generated
View File

File diff suppressed because it is too large Load Diff

81
libs/partners/ibm/pyproject.toml
View File

@ -1,81 +0,0 @@
[build-system]
requires = [ "poetry-core>=1.0.0",]
build-backend = "poetry.core.masonry.api"
[tool.poetry]
name = "langchain-ibm"
version = "0.1.10"
description = "An integration package connecting IBM watsonx.ai and LangChain"
authors = [ "IBM",]
readme = "README.md"
repository = "https://github.com/langchain-ai/langchain"
license = "MIT"
[tool.mypy]
disallow_untyped_defs = "True"
[tool.poetry.urls]
"Source Code" = "https://github.com/langchain-ai/langchain/tree/master/libs/partners/ibm"
[tool.poetry.dependencies]
python = ">=3.10,<4.0"
langchain-core = ">=0.2.17,<0.3"
ibm-watsonx-ai = "^1.0.8"
[tool.ruff.lint]
select = [ "E", "F", "I",]
[tool.coverage.run]
omit = [ "tests/*",]
[tool.pytest.ini_options]
addopts = "--snapshot-warn-unused --strict-markers --strict-config --durations=5"
markers = [ "requires: mark tests as requiring a specific library", "asyncio: mark tests as requiring asyncio", "compile: mark placeholder test used to compile integration tests without running them", "scheduled: mark tests to run in scheduled testing",]
asyncio_mode = "auto"
[tool.poetry.group.test]
optional = true
[tool.poetry.group.codespell]
optional = true
[tool.poetry.group.test_integration]
optional = true
[tool.poetry.group.lint]
optional = true
[tool.poetry.group.dev]
optional = true
[tool.poetry.group.test.dependencies]
pytest = "^7.3.0"
freezegun = "^1.2.2"
pytest-mock = "^3.10.0"
syrupy = "^4.0.2"
pytest-watcher = "^0.3.4"
pytest-asyncio = "^0.21.1"
[tool.poetry.group.codespell.dependencies]
codespell = "^2.2.0"
[tool.poetry.group.test_integration.dependencies]
[tool.poetry.group.lint.dependencies]
ruff = "^0.5"
[tool.poetry.group.typing.dependencies]
mypy = "^1.10"
types-requests = "^2"
[tool.poetry.group.test.dependencies.langchain-core]
path = "../../core"
develop = true
[tool.poetry.group.dev.dependencies.langchain-core]
path = "../../core"
develop = true
[tool.poetry.group.typing.dependencies.langchain-core]
path = "../../core"
develop = true

17
libs/partners/ibm/scripts/check_imports.py
View File

@ -1,17 +0,0 @@
import sys
import traceback
from importlib.machinery import SourceFileLoader
if __name__ == "__main__":
files = sys.argv[1:]
has_failure = False
for file in files:
try:
SourceFileLoader("x", file).load_module()
except Exception:
has_faillure = True
print(file)
traceback.print_exc()
print()
sys.exit(1 if has_failure else 0)

27
libs/partners/ibm/scripts/check_pydantic.sh
View File

@ -1,27 +0,0 @@
#!/bin/bash
#
# This script searches for lines starting with "import pydantic" or "from pydantic"
# in tracked files within a Git repository.
#
# Usage: ./scripts/check_pydantic.sh /path/to/repository
# Check if a path argument is provided
if [ $# -ne 1 ]; then
echo "Usage: $0 /path/to/repository"
exit 1
fi
repository_path="$1"
# Search for lines matching the pattern within the specified repository
result=$(git -C "$repository_path" grep -E '^import pydantic|^from pydantic')
# Check if any matching lines were found
if [ -n "$result" ]; then
echo "ERROR: The following lines need to be updated:"
echo "$result"
echo "Please replace the code with an import from langchain_core.pydantic_v1."
echo "For example, replace 'from pydantic import BaseModel'"
echo "with 'from langchain_core.pydantic_v1 import BaseModel'"
exit 1
fi

17
libs/partners/ibm/scripts/lint_imports.sh
View File

@ -1,17 +0,0 @@
#!/bin/bash
set -eu
# Initialize a variable to keep track of errors
errors=0
# make sure not importing from langchain or langchain_experimental
git --no-pager grep '^from langchain\.' . && errors=$((errors+1))
git --no-pager grep '^from langchain_experimental\.' . && errors=$((errors+1))
# Decide on an exit status based on the errors
if [ "$errors" -gt 0 ]; then
exit 1
else
exit 0
fi

0
libs/partners/ibm/tests/__init__.py
View File

0
libs/partners/ibm/tests/integration_tests/__init__.py
View File

243
libs/partners/ibm/tests/integration_tests/test_chat_models.py
View File

@ -1,243 +0,0 @@
import json
import os
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames # type: ignore
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
SystemMessage,
)
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.pydantic_v1 import BaseModel
from langchain_ibm import ChatWatsonx
WX_APIKEY = os.environ.get("WATSONX_APIKEY", "")
WX_PROJECT_ID = os.environ.get("WATSONX_PROJECT_ID", "")
URL = "https://us-south.ml.cloud.ibm.com"
MODEL_ID = "mistralai/mixtral-8x7b-instruct-v01"
def test_01_generate_chat() -> None:
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
messages = [
("system", "You are a helpful assistant that translates English to French."),
(
"human",
"Translate this sentence from English to French. I love programming.",
),
]
response = chat.invoke(messages)
assert response
def test_01a_generate_chat_with_invoke_params() -> None:
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
params = {
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
messages = [
("system", "You are a helpful assistant that translates English to French."),
(
"human",
"Translate this sentence from English to French. I love programming.",
),
]
response = chat.invoke(messages, params=params)
assert response
def test_01b_generate_chat_with_invoke_params() -> None:
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
params_1 = {
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
params_2 = {
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
chat = ChatWatsonx(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=params_1, # type: ignore[arg-type]
)
messages = [
("system", "You are a helpful assistant that translates English to French."),
(
"human",
"Translate this sentence from English to French. I love programming.",
),
]
response = chat.invoke(messages, params=params_2)
assert response
def test_02_generate_chat_with_few_inputs() -> None:
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
message = HumanMessage(content="Hello")
response = chat.generate([[message], [message]])
assert response
def test_03_generate_chat_with_few_various_inputs() -> None:
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
system_message = SystemMessage(content="You are to chat with the user.")
human_message = HumanMessage(content="Hello")
response = chat.invoke([system_message, human_message])
assert isinstance(response, BaseMessage)
assert isinstance(response.content, str)
def test_05_generate_chat_with_stream() -> None:
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
response = chat.stream("What's the weather in san francisco")
for chunk in response:
assert isinstance(chunk.content, str)
def test_05a_generate_chat_with_stream_with_param() -> None:
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
params = {
GenTextParamsMetaNames.MIN_NEW_TOKENS: 1,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
response = chat.stream("What's the weather in san francisco", params=params)
for chunk in response:
assert isinstance(chunk.content, str)
def test_10_chaining() -> None:
chat = ChatWatsonx(model_id=MODEL_ID, url=URL, project_id=WX_PROJECT_ID) # type: ignore[arg-type]
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant that "
"translates {input_language} to {output_language}.",
),
("human", "{input}"),
]
)
chain = prompt | chat
response = chain.invoke(
{
"input_language": "English",
"output_language": "German",
"input": "I love programming.",
}
)
assert response
def test_11_chaining_with_params() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MIN_NEW_TOKENS: 5,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
chat = ChatWatsonx(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=parameters, # type: ignore[arg-type]
)
prompt = ChatPromptTemplate.from_messages(
[
(
"system",
"You are a helpful assistant that translates "
"{input_language} to {output_language}.",
),
("human", "{input}"),
]
)
chain = prompt | chat
response = chain.invoke(
{
"input_language": "English",
"output_language": "German",
"input": "I love programming.",
}
)
assert response
def test_20_tool_choice() -> None:
"""Test that tool choice is respected."""
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
params = {GenTextParamsMetaNames.MAX_NEW_TOKENS: 500}
chat = ChatWatsonx(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=params, # type: ignore[arg-type]
)
class MyTool(BaseModel):
name: str
age: int
with_tool = chat.bind_tools([MyTool], tool_choice="MyTool")
resp = with_tool.invoke("Who was the 27 year old named Erick?")
assert isinstance(resp, AIMessage)
assert resp.content == "" # should just be tool call
tool_calls = resp.additional_kwargs["tool_calls"]
assert len(tool_calls) == 1
tool_call = tool_calls[0]
assert tool_call["function"]["name"] == "MyTool"
assert json.loads(tool_call["function"]["arguments"]) == {
"age": 27,
"name": "Erick",
}
assert tool_call["type"] == "function"
assert isinstance(resp.tool_calls, list)
assert len(resp.tool_calls) == 1
tool_call = resp.tool_calls[0]
assert tool_call["name"] == "MyTool"
assert tool_call["args"] == {"age": 27, "name": "Erick"}
def test_21_tool_choice_bool() -> None:
"""Test that tool choice is respected just passing in True."""
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames
params = {GenTextParamsMetaNames.MAX_NEW_TOKENS: 500}
chat = ChatWatsonx(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=params, # type: ignore[arg-type]
)
class MyTool(BaseModel):
name: str
age: int
with_tool = chat.bind_tools([MyTool], tool_choice=True)
resp = with_tool.invoke("Who was the 27 year old named Erick?")
assert isinstance(resp, AIMessage)
assert resp.content == "" # should just be tool call
tool_calls = resp.additional_kwargs["tool_calls"]
assert len(tool_calls) == 1
tool_call = tool_calls[0]
assert tool_call["function"]["name"] == "MyTool"
assert json.loads(tool_call["function"]["arguments"]) == {
"age": 27,
"name": "Erick",
}
assert tool_call["type"] == "function"

7
libs/partners/ibm/tests/integration_tests/test_compile.py
View File

@ -1,7 +0,0 @@
import pytest # type: ignore[import-not-found]
@pytest.mark.compile
def test_placeholder() -> None:
"""Used for compiling integration tests without running any real tests."""
pass

73
libs/partners/ibm/tests/integration_tests/test_embeddings.py
View File

@ -1,73 +0,0 @@
"""Test WatsonxEmbeddings.
You'll need to set WATSONX_APIKEY and WATSONX_PROJECT_ID environment variables.
"""
import os
from ibm_watsonx_ai import APIClient # type: ignore
from ibm_watsonx_ai.metanames import EmbedTextParamsMetaNames # type: ignore
from langchain_ibm import WatsonxEmbeddings
WX_APIKEY = os.environ.get("WATSONX_APIKEY", "")
WX_PROJECT_ID = os.environ.get("WATSONX_PROJECT_ID", "")
URL = "https://us-south.ml.cloud.ibm.com"
MODEL_ID = "ibm/slate-125m-english-rtrvr"
DOCUMENTS = ["What is a generative ai?", "What is a loan and how does it works?"]
def test_01_generate_embed_documents() -> None:
watsonx_embedding = WatsonxEmbeddings(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID, # type: ignore[arg-type]
)
generate_embedding = watsonx_embedding.embed_documents(texts=DOCUMENTS)
assert len(generate_embedding) == len(DOCUMENTS)
assert all(isinstance(el, float) for el in generate_embedding[0])
def test_02_generate_embed_query() -> None:
watsonx_embedding = WatsonxEmbeddings(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
generate_embedding = watsonx_embedding.embed_query(text=DOCUMENTS[0])
assert isinstance(generate_embedding, list) and isinstance(
generate_embedding[0], float
)
def test_03_generate_embed_documents_with_param() -> None:
embed_params = {
EmbedTextParamsMetaNames.TRUNCATE_INPUT_TOKENS: 3,
}
watsonx_embedding = WatsonxEmbeddings(
model_id=MODEL_ID,
url=URL, # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=embed_params, # type: ignore[arg-type]
)
generate_embedding = watsonx_embedding.embed_documents(texts=DOCUMENTS)
assert len(generate_embedding) == len(DOCUMENTS)
assert all(isinstance(el, float) for el in generate_embedding[0])
def test_10_generate_embed_query_with_client_initialization() -> None:
watsonx_client = APIClient(
credentials={
"url": URL,
"apikey": WX_APIKEY,
}
)
watsonx_embedding = WatsonxEmbeddings(
model_id=MODEL_ID, project_id=WX_PROJECT_ID, watsonx_client=watsonx_client
)
generate_embedding = watsonx_embedding.embed_query(text=DOCUMENTS[0])
assert isinstance(generate_embedding, list) and isinstance(
generate_embedding[0], float
)

292
libs/partners/ibm/tests/integration_tests/test_llms.py
View File

@ -1,292 +0,0 @@
"""Test WatsonxLLM API wrapper.
You'll need to set WATSONX_APIKEY and WATSONX_PROJECT_ID environment variables.
"""
import os
from ibm_watsonx_ai import Credentials # type: ignore
from ibm_watsonx_ai.foundation_models import Model, ModelInference # type: ignore
from ibm_watsonx_ai.foundation_models.utils.enums import ( # type: ignore
DecodingMethods,
ModelTypes,
)
from ibm_watsonx_ai.metanames import GenTextParamsMetaNames # type: ignore
from langchain_core.outputs import LLMResult
from langchain_ibm import WatsonxLLM
WX_APIKEY = os.environ.get("WATSONX_APIKEY", "")
WX_PROJECT_ID = os.environ.get("WATSONX_PROJECT_ID", "")
MODEL_ID = "google/flan-ul2"
def test_watsonxllm_invoke() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_with_params() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 5,
}
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=parameters,
)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_with_params_2() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 5,
}
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.invoke("What color sunflower is?", params=parameters)
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_with_params_3() -> None:
parameters_1 = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
}
parameters_2 = {
GenTextParamsMetaNames.MIN_NEW_TOKENS: 5,
}
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
params=parameters_1,
)
response = watsonxllm.invoke("What color sunflower is?", params=parameters_2)
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_generate() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.generate(["What color sunflower is?"])
print(f"\nResponse: {response}")
response_text = response.generations[0][0].text
print(f"Response text: {response_text}")
assert isinstance(response, LLMResult)
assert len(response_text) > 0
def test_watsonxllm_generate_with_param() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 10,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 5,
}
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.generate(["What color sunflower is?"], params=parameters)
print(f"\nResponse: {response}")
response_text = response.generations[0][0].text
print(f"Response text: {response_text}")
assert isinstance(response, LLMResult)
assert len(response_text) > 0
def test_watsonxllm_generate_with_multiple_prompts() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.generate(
["What color sunflower is?", "What color turtle is?"]
)
print(f"\nResponse: {response}")
response_text = response.generations[0][0].text
print(f"Response text: {response_text}")
assert isinstance(response, LLMResult)
assert len(response_text) > 0
def test_watsonxllm_generate_stream() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.generate(["What color sunflower is?"], stream=True)
print(f"\nResponse: {response}")
response_text = response.generations[0][0].text
print(f"Response text: {response_text}")
assert isinstance(response, LLMResult)
assert len(response_text) > 0
def test_watsonxllm_stream() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
stream_response = watsonxllm.stream("What color sunflower is?")
linked_text_stream = ""
for chunk in stream_response:
assert isinstance(
chunk, str
), f"chunk expect type '{str}', actual '{type(chunk)}'"
linked_text_stream += chunk
print(f"Linked text stream: {linked_text_stream}")
assert (
response == linked_text_stream
), "Linked text stream are not the same as generated text"
def test_watsonxllm_invoke_from_wx_model() -> None:
model = Model(
model_id=MODEL_ID,
credentials={
"apikey": WX_APIKEY,
"url": "https://us-south.ml.cloud.ibm.com",
},
project_id=WX_PROJECT_ID,
)
watsonxllm = WatsonxLLM(watsonx_model=model)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_from_wx_model_inference() -> None:
credentials = Credentials(
api_key=WX_APIKEY, url="https://us-south.ml.cloud.ibm.com"
)
model = ModelInference(
model_id=MODEL_ID,
credentials=credentials,
project_id=WX_PROJECT_ID,
)
watsonxllm = WatsonxLLM(watsonx_model=model)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_from_wx_model_inference_with_params() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: "sample",
GenTextParamsMetaNames.MAX_NEW_TOKENS: 100,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 10,
GenTextParamsMetaNames.TEMPERATURE: 0.5,
GenTextParamsMetaNames.TOP_K: 50,
GenTextParamsMetaNames.TOP_P: 1,
}
model = ModelInference(
model_id=MODEL_ID,
credentials={
"apikey": WX_APIKEY,
"url": "https://us-south.ml.cloud.ibm.com",
},
project_id=WX_PROJECT_ID,
params=parameters,
)
watsonxllm = WatsonxLLM(watsonx_model=model)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
def test_watsonxllm_invoke_from_wx_model_inference_with_params_as_enum() -> None:
parameters = {
GenTextParamsMetaNames.DECODING_METHOD: DecodingMethods.GREEDY,
GenTextParamsMetaNames.MAX_NEW_TOKENS: 100,
GenTextParamsMetaNames.MIN_NEW_TOKENS: 10,
GenTextParamsMetaNames.TEMPERATURE: 0.5,
GenTextParamsMetaNames.TOP_K: 50,
GenTextParamsMetaNames.TOP_P: 1,
}
model = ModelInference(
model_id=ModelTypes.FLAN_UL2,
credentials={
"apikey": WX_APIKEY,
"url": "https://us-south.ml.cloud.ibm.com",
},
project_id=WX_PROJECT_ID,
params=parameters,
)
watsonxllm = WatsonxLLM(watsonx_model=model)
response = watsonxllm.invoke("What color sunflower is?")
print(f"\nResponse: {response}")
assert isinstance(response, str)
assert len(response) > 0
async def test_watsonx_ainvoke() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = await watsonxllm.ainvoke("What color sunflower is?")
assert isinstance(response, str)
async def test_watsonx_agenerate() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
response = await watsonxllm.agenerate(
["What color sunflower is?", "What color turtle is?"]
)
assert len(response.generations) > 0
assert response.llm_output["token_usage"]["generated_token_count"] != 0 # type: ignore
def test_get_num_tokens() -> None:
watsonxllm = WatsonxLLM(
model_id=MODEL_ID,
url="https://us-south.ml.cloud.ibm.com", # type: ignore[arg-type]
project_id=WX_PROJECT_ID,
)
num_tokens = watsonxllm.get_num_tokens("What color sunflower is?")
assert num_tokens > 0

0
libs/partners/ibm/tests/unit_tests/__init__.py
View File

62
libs/partners/ibm/tests/unit_tests/test_chat_models.py
View File

@ -1,62 +0,0 @@
"""Test ChatWatsonx API wrapper."""
import os
from langchain_ibm import ChatWatsonx
os.environ.pop("WATSONX_APIKEY", None)
os.environ.pop("WATSONX_PROJECT_ID", None)
MODEL_ID = "mistralai/mixtral-8x7b-instruct-v01"
def test_initialize_chat_watsonx_bad_path_without_url() -> None:
try:
ChatWatsonx(
model_id=MODEL_ID,
)
except ValueError as e:
assert "WATSONX_URL" in e.__str__()
def test_initialize_chat_watsonx_cloud_bad_path() -> None:
try:
ChatWatsonx(model_id=MODEL_ID, url="https://us-south.ml.cloud.ibm.com") # type: ignore[arg-type]
except ValueError as e:
assert "WATSONX_APIKEY" in e.__str__()
def test_initialize_chat_watsonx_cpd_bad_path_without_all() -> None:
try:
ChatWatsonx(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
)
except ValueError as e:
assert (
"WATSONX_APIKEY" in e.__str__()
and "WATSONX_PASSWORD" in e.__str__()
and "WATSONX_TOKEN" in e.__str__()
)
def test_initialize_chat_watsonx_cpd_bad_path_password_without_username() -> None:
try:
ChatWatsonx(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
password="test_password", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()
def test_initialize_chat_watsonx_cpd_bad_path_apikey_without_username() -> None:
try:
ChatWatsonx(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
apikey="test_apikey", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()

62
libs/partners/ibm/tests/unit_tests/test_embeddings.py
View File

@ -1,62 +0,0 @@
"""Test WatsonxLLM API wrapper."""
import os
from langchain_ibm import WatsonxEmbeddings
os.environ.pop("WATSONX_APIKEY", None)
os.environ.pop("WATSONX_PROJECT_ID", None)
MODEL_ID = "ibm/slate-125m-english-rtrvr"
def test_initialize_watsonx_embeddings_bad_path_without_url() -> None:
try:
WatsonxEmbeddings(
model_id=MODEL_ID,
)
except ValueError as e:
assert "WATSONX_URL" in e.__str__()
def test_initialize_watsonx_embeddings_cloud_bad_path() -> None:
try:
WatsonxEmbeddings(model_id=MODEL_ID, url="https://us-south.ml.cloud.ibm.com") # type: ignore[arg-type]
except ValueError as e:
assert "WATSONX_APIKEY" in e.__str__()
def test_initialize_watsonx_embeddings_cpd_bad_path_without_all() -> None:
try:
WatsonxEmbeddings(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
)
except ValueError as e:
assert (
"WATSONX_APIKEY" in e.__str__()
and "WATSONX_PASSWORD" in e.__str__()
and "WATSONX_TOKEN" in e.__str__()
)
def test_initialize_watsonx_embeddings_cpd_bad_path_password_without_username() -> None:
try:
WatsonxEmbeddings(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
password="test_password", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()
def test_initialize_watsonx_embeddings_cpd_bad_path_apikey_without_username() -> None:
try:
WatsonxEmbeddings(
model_id=MODEL_ID,
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
apikey="test_apikey", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()

7
libs/partners/ibm/tests/unit_tests/test_imports.py
View File

@ -1,7 +0,0 @@
from langchain_ibm import __all__
EXPECTED_ALL = ["WatsonxLLM", "WatsonxEmbeddings", "ChatWatsonx"]
def test_all_imports() -> None:
assert sorted(EXPECTED_ALL) == sorted(__all__)

60
libs/partners/ibm/tests/unit_tests/test_llms.py
View File

@ -1,60 +0,0 @@
"""Test WatsonxLLM API wrapper."""
import os
from langchain_ibm import WatsonxLLM
os.environ.pop("WATSONX_APIKEY", None)
os.environ.pop("WATSONX_PROJECT_ID", None)
def test_initialize_watsonxllm_bad_path_without_url() -> None:
try:
WatsonxLLM(
model_id="google/flan-ul2",
)
except ValueError as e:
assert "WATSONX_URL" in e.__str__()
def test_initialize_watsonxllm_cloud_bad_path() -> None:
try:
WatsonxLLM(model_id="google/flan-ul2", url="https://us-south.ml.cloud.ibm.com") # type: ignore[arg-type]
except ValueError as e:
assert "WATSONX_APIKEY" in e.__str__()
def test_initialize_watsonxllm_cpd_bad_path_without_all() -> None:
try:
WatsonxLLM(
model_id="google/flan-ul2",
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
)
except ValueError as e:
assert (
"WATSONX_APIKEY" in e.__str__()
and "WATSONX_PASSWORD" in e.__str__()
and "WATSONX_TOKEN" in e.__str__()
)
def test_initialize_watsonxllm_cpd_bad_path_password_without_username() -> None:
try:
WatsonxLLM(
model_id="google/flan-ul2",
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
password="test_password", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()
def test_initialize_watsonxllm_cpd_bad_path_apikey_without_username() -> None:
try:
WatsonxLLM(
model_id="google/flan-ul2",
url="https://cpd-zen.apps.cpd48.cp.fyre.ibm.com", # type: ignore[arg-type]
apikey="test_apikey", # type: ignore[arg-type]
)
except ValueError as e:
assert "WATSONX_USERNAME" in e.__str__()