Add bind tools to custom BaseChatModel #26146
Replies: 3 comments 9 replies
-
|
To bind tools to your custom
Here's how you can implement it: from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.utils.function_calling import convert_to_openai_tool
# Define your tools
class GetWeather(BaseModel):
'''Get the current weather in a given location'''
location: str = Field(..., description="The city and state, e.g. San Francisco, CA")
class GetPopulation(BaseModel):
'''Get the current population in a given location'''
location: str = Field(..., description="The city and state, e.g. San Francisco, CA")
# Your custom chat model
class GPT4oCustomChatModel(BaseChatModel):
# ... (your existing implementation)
def bind_tools(self, tools):
formatted_tools = [convert_to_openai_tool(tool) for tool in tools]
return super().bind_tools(formatted_tools)
# Initialize your custom chat model
llm = GPT4oCustomChatModel(
client_id="your_client_id",
client_secret="your_client_secret",
apim_Subscription_Key="your_subscription_key",
api_url="https://api.yourservice.com/v1",
model_name="gpt-4o",
token_manager=TokenManager(),
temperature=0.7,
top_p=0.95,
max_tokens=8000,
)
# Bind the tools to the chat model
tools = [GetWeather, GetPopulation]
llm_with_tools = llm.bind_tools(tools)
# Example usage
ai_msg = llm_with_tools.invoke("Which city is hotter today and which is bigger: LA or NY?")
print(ai_msg.tool_calls)This code defines two tools, For more advanced use cases, such as creating agents that can interact with multiple tools and handle more complex workflows, you might want to explore the |
Beta Was this translation helpful? Give feedback.
-
|
with your last proposed code, the following error is generated [2024-09-06T13:59:06.033Z] Executed 'Functions.AsistenteDigital' (Failed, Id=364f7583-d1e6-4f5f-a560-d0a097784286, Duration=12170ms) |
Beta Was this translation helpful? Give feedback.
-
|
To resolve the from typing import Any, Dict, List, Optional, Sequence, Union, Callable, Type
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage, SystemMessage
from langchain_core.outputs import ChatResult, ChatGeneration
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_core.runnable import Runnable
from langchain_core.pydantic_v1 import BaseModel
from models.token_manager import TokenManager
class GPT4oCustomChatModel(BaseChatModel):
"""Custom chat model that invokes GPT-4o from a given endpoint."""
client_id: str
client_secret: str
apim_Subscription_Key: str
api_url: str
model_name: str
token_manager: TokenManager
temperature: float = 0.7
top_p: float = 0.95
max_tokens: int = 8000
class Config:
arbitrary_types_allowed = True
def _get_valid_token(self) -> str:
"""Obtiene un token válido. Si ha expirado, genera uno nuevo."""
if not self.token_manager.chequear_tiempo_vida_token():
return self.token_manager.generar_token(self.client_id, self.client_secret)
return self.token_manager.token
def _format_messages(self, messages: List[BaseMessage]) -> List[Dict[str, Any]]:
"""Format LangChain messages to the API format."""
return [
{
"role": "system" if isinstance(msg, SystemMessage)
else "assistant" if isinstance(msg, AIMessage)
else "user",
"content": [{"type": "text", "text": msg.content}]
} for msg in messages
]
def _prepare_payload(self, messages: List[BaseMessage]) -> Dict[str, Any]:
"""Prepare the payload for the API request."""
return {
"messages": self._format_messages(messages),
"temperature": self.temperature,
"top_p": self.top_p,
"max_tokens": self.max_tokens,
}
def _process_response(self, response_data: Dict[str, Any]) -> ChatResult:
"""Process the response from the API and return a ChatResult."""
choice = response_data["choices"][0]
message_content = choice["message"]["content"]
finish_reason = choice.get("finish_reason", "unknown")
content_filter_error = choice.get(
"content_filter_result", {}).get("error", None)
if content_filter_error:
print(f"Content filter error: {content_filter_error['message']}")
ai_message = AIMessage(content=message_content)
generation = ChatGeneration(
message=ai_message, finish_reason=finish_reason)
return ChatResult(generations=[generation])
def _make_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Make a synchronous HTTP request."""
headers = {"Authorization": f"Bearer {self._get_valid_token()}",
"Ocp-Apim-Subscription-Key": self.apim_Subscription_Key}
response = httpx.post(self.api_url, headers=headers, json=payload)
return response.json()
async def _make_async_request(self, payload: Dict[str, Any]) -> Dict[str, Any]:
"""Make an asynchronous HTTP request."""
headers = {"Authorization": f"Bearer {self._get_valid_token()}",
"Ocp-Apim-Subscription-Key": self.apim_Subscription_Key}
async with httpx.AsyncClient() as client:
response = await client.post(self.api_url, headers=headers, json=payload)
return response.json()
def _generate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> ChatResult:
"""Generates a response from the GPT-4o model by invoking the API."""
payload = self._prepare_payload(messages)
response_data = self._make_request(payload)
return self._process_response(response_data)
async def _agenerate(
self,
messages: List[BaseMessage],
stop: Optional[List[str]] = None,
**kwargs: Any,
) -> ChatResult:
"""Asynchronously generates a response from the GPT-4o model by invoking the API."""
payload = self._prepare_payload(messages)
response_data = await self._make_async_request(payload)
return self._process_response(response_data)
@property
def _llm_type(self) -> str:
"""Returns the type of this model for logging purposes."""
return "gpt-4o-custom-chatmodel"
@property
def _identifying_params(self) -> Dict[str, Any]:
"""Identifying parameters used for tracing."""
return {
"model_name": self.model_name,
"temperature": self.temperature,
"top_p": self.top_p,
"max_tokens": self.max_tokens,
}
def bind_tools(
self,
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
*,
tool_choice: Optional[
Union[dict, str, Literal["auto", "none", "any"], bool]
] = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
"""Bind tool-like objects to this chat model.
Assumes model is compatible with OpenAI tool-calling API.
Args:
tools: A list of tool definitions to bind to this chat model.
Supports any tool definition handled by
:meth:`langchain_core.utils.function_calling.convert_to_openai_tool`.
tool_choice: Which tool 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), or a dict of the form:
{"type": "function", "function": {"name": <<tool_name>>}}.
**kwargs: Any additional parameters to pass to the
:class:`~langchain.runnable.Runnable` constructor.
"""
formatted_tools = [convert_to_openai_tool(tool) for tool in tools]
if tool_choice is not None and tool_choice:
if len(formatted_tools) != 1:
raise ValueError(
"When specifying `tool_choice`, you must provide exactly one "
f"tool. Received {len(formatted_tools)} tools."
)
if isinstance(tool_choice, str):
if tool_choice not in ("auto", "none", "any"):
tool_choice = {
"type": "function",
"function": {"name": tool_choice},
}
elif isinstance(tool_choice, bool):
tool_choice = formatted_tools[0]
elif isinstance(tool_choice, dict):
if (
formatted_tools[0]["function"]["name"]
!= tool_choice["function"]["name"]
):
raise ValueError(
f"Tool choice {tool_choice} was specified, but the only "
f"provided tool was {formatted_tools[0]['function']['name']}."
)
else:
raise ValueError(
f"Unrecognized tool_choice type. Expected str, bool or dict. "
f"Received: {tool_choice}"
)
kwargs["tool_choice"] = tool_choice
return super().bind_tools(tools=formatted_tools, **kwargs)This implementation ensures that the |
Beta Was this translation helpful? Give feedback.
-
|
the problem is that when you reference return super().bind_tools(tools=formatted_tools, **kwargs) you call bind_tools from BaseLanguageModel, and the method in its class is |
Beta Was this translation helpful? Give feedback.
-
|
the problem is that when you reference return super().bind_tools(tools=formatted_tools, **kwargs) you call bind_tools from BaseLanguageModel, and the method in its class is how fix that? |
Beta Was this translation helpful? Give feedback.
-
|
To fix the issue where calling Here is an example of how you can override the from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.tools import BaseTool
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_core.runnables import Runnable
from langchain_core.language_models import LanguageModelInput
from typing import Sequence, Union, Dict, Any, Type, Callable, Optional, Literal
class GPT4oCustomChatModel(BaseChatModel):
def bind_tools(
self,
tools: Sequence[Union[Dict[str, Any], Type[BaseModel], Callable, BaseTool]],
*,
tool_choice: Optional[
Union[dict, str, Literal["auto", "none", "required", "any"], bool]
] = None,
strict: Optional[bool] = None,
**kwargs: Any,
) -> Runnable[LanguageModelInput, BaseMessage]:
"""Bind tool-like objects to this chat model.
Assumes model is compatible with OpenAI tool-calling API.
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.
tool_choice: Specifies how to choose the tool to call.
strict: Whether to enforce strict validation of tools.
**kwargs: Any additional parameters to pass to the
:class:`~langchain.runnable.Runnable` constructor.
"""
formatted_tools = [convert_to_openai_tool(tool, strict=strict) for tool in tools]
if tool_choice:
if isinstance(tool_choice, str):
if tool_choice not in ("auto", "none", "any", "required"):
tool_choice = {
"type": "function",
"function": {"name": tool_choice},
}
if tool_choice == "any":
tool_choice = "required"
elif isinstance(tool_choice, bool):
tool_choice = "required"
elif isinstance(tool_choice, dict):
tool_names = [
formatted_tool["function"]["name"]
for formatted_tool in formatted_tools
]
if not any(
tool_name == tool_choice["function"]["name"]
for tool_name in tool_names
):
raise ValueError(
f"Tool choice {tool_choice} was specified, but the only "
f"provided tools were {tool_names}."
)
else:
raise ValueError(
f"Unrecognized tool_choice type. Expected str, bool or dict. "
f"Received: {tool_choice}"
)
kwargs["tool_choice"] = tool_choice
# Implement the binding logic here
# For example, you can store the formatted tools in an instance variable
self._formatted_tools = formatted_tools
# Return a Runnable instance or perform other necessary actions
return Runnable(formatted_tools, tool_choice=tool_choice, **kwargs)In this example, the |
Beta Was this translation helpful? Give feedback.
-
|
Hi @santiagovasquez1 , Facing same issue, can you please let me know how you solved this problem. but my customchatmodel class is not able to recognize presence of any tool. Any thoughts please ? |
Beta Was this translation helpful? Give feedback.
-
|
you can find example of how to implement a BaseChatModel in langchain package BaseChatOpenAI. |
Beta Was this translation helpful? Give feedback.
-
|
Hi Santiago, how is it going? Is your custom chat model finally well implemented for the tool binding? Is it public anywhere? Im getting the same issues as you, thanks in advance! |
Beta Was this translation helpful? Give feedback.
-
|
I was able to solve the use of tools for my custom model as follows. def _prepare_payload(self, messages: List[BaseMessage]) -> Dict[str, Any]:
"""Prepares the payload for the API request.
Args:
messages (List[BaseMessage]): List of messages to include in the payload.
Returns:
Dict[str, Any]: The prepared payload.
"""
payload = {
"messages": self._format_messages(messages),
"temperature": self.temperature,
"top_p": self.top_p,
"max_tokens": self.max_tokens,
}
if self.tools:
payload["tools"] = self.tools
self.is_structured_output = False
self.tools = None
return payload
def _process_tool_calls(
self, tool_calls_data: List[Dict[str, Any]]
) -> List[ToolCall]:
"""Processes tool call data and returns a list of ToolCall objects.
Args:
tool_calls_data (List[Dict[str, Any]]): Data for tool calls.
Returns:
List[ToolCall]: List of ToolCall objects.
"""
tool_calls = []
for tool_call in tool_calls_data:
tool_calls.append(
ToolCall(
name=tool_call["function"]["name"],
args=json.loads(tool_call["function"]["arguments"]),
id=tool_call["id"],
)
)
return tool_calls
def _process_response(self, response_data: Dict[str, Any]) -> ChatResult:
"""Processes the response from the API and returns a ChatResult.
Args:
response_data (Dict[str, Any]): The response data from the API.
Returns:
ChatResult: The processed chat result.
"""
choice = response_data["choices"][0]
finish_reason = choice.get("finish_reason", "unknown")
if finish_reason == "tool_calls" and "tool_calls" in choice["message"]:
tool_calls = self._process_tool_calls(choice["message"]["tool_calls"])
ai_message = AIMessage(content="", tool_calls=tool_calls)
else:
ai_message = AIMessage(content=choice["message"].get("content", ""))
if content_filter_error := choice.get("content_filter_results", {}).get(
"error"
):
logging.warning(f"Content filter error: {content_filter_error['message']}")
return ChatResult(
generations=[
ChatGeneration(message=ai_message, finish_reason=finish_reason)
]
)
def with_structured_output(
self,
schema: Optional[Any] = None,
*,
method: Literal["function_calling", "json_mode"] = "function_calling",
**kwargs: Any,
) -> Union["GPT4oCustomChatModel", RunnableSerializable[LanguageModelInput, Any]]:
"""Enables structured output based on schema and method provided.
Args:
schema (Optional[Any]): The schema for structured output.
method (Literal["function_calling", "json_mode"]): The method for structured output.
**kwargs (Any): Additional arguments.
Returns:
Union[GPT4oCustomChatModel, RunnableSerializable[LanguageModelInput, Any]]: The model with structured output enabled.
"""
if kwargs:
raise ValueError(f"Received unsupported arguments {kwargs}")
# Validar el método
if method not in ["function_calling", "json_mode", "json_schema"]:
raise ValueError(f"Unrecognized method argument. Received: '{method}'")
is_pydantic_schema = is_basemodel_subclass(schema)
model_copy = deepcopy(self)
# Inicializamos las variables
model_copy.tools = []
output_parser = None
# Preparar la herramienta adecuada según el método
if method == "function_calling":
if schema is None:
raise ValueError(
"A schema must be provided when using 'function_calling'"
)
tool = convert_to_openai_tool(schema)
model_copy.tools = [tool]
model_copy.is_structured_output = True
# Generar el parser correspondiente
if is_pydantic_schema:
output_parser = PydanticToolsParser(
tools=[schema], first_tool_only=True
)
else:
output_parser = JsonOutputKeyToolsParser(
key_name=tool["function"]["name"], first_tool_only=True
)
elif method == "json_mode":
model_copy.is_structured_output = True
# Asignar parser basado en si es un esquema de Pydantic o no
output_parser = (
PydanticOutputParser(pydantic_object=schema)
if is_pydantic_schema
else JsonOutputParser()
)
return model_copy | output_parser if output_parser else model_copy
|
Beta Was this translation helpful? Give feedback.
-
Checked other resources
Commit to Help
Example Code
Description
I have a custom BaseChatModel that call a gpt 4.o mini via a rest api via url I need bind tools to it, how can I do that?
System Info
langchain==0.2.14
langchain-cli==0.0.30
langchain-community==0.2.12
langchain-core==0.2.35
langchain-experimental==0.0.64
langchain-openai==0.1.22
langchain-text-splitters==0.2.2
Beta Was this translation helpful? Give feedback.
All reactions