@ -1,74 +1,31 @@
import json
from abc import ABC , abstractmethod
from typing import Type
import pytest
from langchain_core . language_models import BaseChatModel
from langchain_core . messages import AIMessage , AIMessageChunk , HumanMessage , ToolMessage
from langchain_core . pydantic_v1 import BaseModel , Field
from langchain_core . tools import tool
from langchain_standard_tests . unit_tests . chat_models import (
ChatModelTests ,
my_adder_tool ,
)
class Person ( BaseModel ) :
name : str = Field ( . . . , description = " The name of the person. " )
age : int = Field ( . . . , description = " The age of the person. " )
@tool
def my_adder_tool ( a : int , b : int ) - > int :
""" Takes two integers, a and b, and returns their sum. """
return a + b
class ChatModelIntegrationTests ( ABC ) :
@abstractmethod
@pytest.fixture
def chat_model_class ( self ) - > Type [ BaseChatModel ] :
. . .
@pytest.fixture
def chat_model_params ( self ) - > dict :
return { }
@pytest.fixture
def chat_model_has_tool_calling (
self , chat_model_class : Type [ BaseChatModel ]
) - > bool :
return chat_model_class . bind_tools is not BaseChatModel . bind_tools
@pytest.fixture
def chat_model_has_structured_output (
self , chat_model_class : Type [ BaseChatModel ]
) - > bool :
return (
chat_model_class . with_structured_output
is not BaseChatModel . with_structured_output
)
def test_invoke (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
class ChatModelIntegrationTests ( ChatModelTests ) :
def test_invoke ( self , model : BaseChatModel ) - > None :
result = model . invoke ( " Hello " )
assert result is not None
assert isinstance ( result , AIMessage )
assert isinstance ( result . content , str )
assert len ( result . content ) > 0
async def test_ainvoke (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
async def test_ainvoke ( self , model : BaseChatModel ) - > None :
result = await model . ainvoke ( " Hello " )
assert result is not None
assert isinstance ( result , AIMessage )
assert isinstance ( result . content , str )
assert len ( result . content ) > 0
def test_stream (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
def test_stream ( self , model : BaseChatModel ) - > None :
num_tokens = 0
for token in model . stream ( " Hello " ) :
assert token is not None
@ -76,10 +33,7 @@ class ChatModelIntegrationTests(ABC):
num_tokens + = len ( token . content )
assert num_tokens > 0
async def test_astream (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
async def test_astream ( self , model : BaseChatModel ) - > None :
num_tokens = 0
async for token in model . astream ( " Hello " ) :
assert token is not None
@ -87,10 +41,7 @@ class ChatModelIntegrationTests(ABC):
num_tokens + = len ( token . content )
assert num_tokens > 0
def test_batch (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
def test_batch ( self , model : BaseChatModel ) - > None :
batch_results = model . batch ( [ " Hello " , " Hey " ] )
assert batch_results is not None
assert isinstance ( batch_results , list )
@ -101,10 +52,7 @@ class ChatModelIntegrationTests(ABC):
assert isinstance ( result . content , str )
assert len ( result . content ) > 0
async def test_abatch (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
async def test_abatch ( self , model : BaseChatModel ) - > None :
batch_results = await model . abatch ( [ " Hello " , " Hey " ] )
assert batch_results is not None
assert isinstance ( batch_results , list )
@ -115,14 +63,11 @@ class ChatModelIntegrationTests(ABC):
assert isinstance ( result . content , str )
assert len ( result . content ) > 0
def test_conversation (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
def test_conversation ( self , model : BaseChatModel ) - > None :
messages = [
HumanMessage ( content = " hello " ) ,
AIMessage ( content = " hello " ) ,
HumanMessage ( content = " how are you " ) ,
HumanMessage ( " hello " ) ,
AIMessage ( " hello " ) ,
HumanMessage ( " how are you " ) ,
]
result = model . invoke ( messages )
assert result is not None
@ -130,10 +75,9 @@ class ChatModelIntegrationTests(ABC):
assert isinstance ( result . content , str )
assert len ( result . content ) > 0
def test_usage_metadata (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
def test_usage_metadata ( self , model : BaseChatModel ) - > None :
if not self . returns_usage_metadata :
pytest . skip ( " Not implemented. " )
result = model . invoke ( " Hello " )
assert result is not None
assert isinstance ( result , AIMessage )
@ -142,39 +86,35 @@ class ChatModelIntegrationTests(ABC):
assert isinstance ( result . usage_metadata [ " output_tokens " ] , int )
assert isinstance ( result . usage_metadata [ " total_tokens " ] , int )
def test_stop_sequence (
self , chat_model_class : Type [ BaseChatModel ] , chat_model_params : dict
) - > None :
model = chat_model_class ( * * chat_model_params )
def test_stop_sequence ( self , model : BaseChatModel ) - > None :
result = model . invoke ( " hi " , stop = [ " you " ] )
assert isinstance ( result , AIMessage )
model = chat_model_class ( * * chat_model_params , stop = [ " you " ] )
result = model . invoke ( " hi " )
custom_model = self . chat_model_class (
* * { * * self . chat_model_params , " stop " : [ " you " ] }
)
result = custom_model . invoke ( " hi " )
assert isinstance ( result , AIMessage )
def test_tool_message_histories_string_content (
self ,
chat_model_class : Type [ BaseChatModel ] ,
chat_model_params : dict ,
chat_model_has_tool_calling : bool ,
model : BaseChatModel ,
) - > None :
"""
Test that message histories are compatible with string tool contents
( e . g . OpenAI ) .
"""
if not chat_model_ :
if not self . :
pytest . skip ( " Test requires tool calling. " )
model = chat_model_class ( * * chat_model_params )
model_with_tools = model . bind_tools ( [ my_adder_tool ] )
function_name = " my_adder_tool "
function_args = { " a " : " 1 " , " b " : " 2 " }
messages_string_content = [
HumanMessage ( content = " What is 1 + 2 " ) ,
HumanMessage ( " What is 1 + 2 " ) ,
# string content (e.g. OpenAI)
AIMessage (
content = " " ,
" " ,
tool_calls = [
{
" name " : function_name ,
@ -184,8 +124,8 @@ class ChatModelIntegrationTests(ABC):
] ,
) ,
ToolMessage (
json . dumps ( { " result " : 3 } ) ,
name = function_name ,
content = json . dumps ( { " result " : 3 } ) ,
tool_call_id = " abc123 " ,
) ,
]
@ -194,26 +134,23 @@ class ChatModelIntegrationTests(ABC):
def test_tool_message_histories_list_content (
self ,
chat_model_class : Type [ BaseChatModel ] ,
chat_model_params : dict ,
chat_model_has_tool_calling : bool ,
model : BaseChatModel ,
) - > None :
"""
Test that message histories are compatible with list tool contents
( e . g . Anthropic ) .
"""
if not chat_model_ :
if not self . :
pytest . skip ( " Test requires tool calling. " )
model = chat_model_class ( * * chat_model_params )
model_with_tools = model . bind_tools ( [ my_adder_tool ] )
function_name = " my_adder_tool "
function_args = { " a " : 1 , " b " : 2 }
messages_list_content = [
HumanMessage ( content = " What is 1 + 2 " ) ,
HumanMessage ( " What is 1 + 2 " ) ,
# List content (e.g., Anthropic)
AIMessage (
content = [
[
{ " type " : " text " , " text " : " some text " } ,
{
" type " : " tool_use " ,
@ -231,8 +168,8 @@ class ChatModelIntegrationTests(ABC):
] ,
) ,
ToolMessage (
json . dumps ( { " result " : 3 } ) ,
name = function_name ,
content = json . dumps ( { " result " : 3 } ) ,
tool_call_id = " abc123 " ,
) ,
]
@ -241,25 +178,22 @@ class ChatModelIntegrationTests(ABC):
def test_structured_few_shot_examples (
self ,
chat_model_class : Type [ BaseChatModel ] ,
chat_model_params : dict ,
chat_model_has_tool_calling : bool ,
model : BaseChatModel ,
) - > None :
"""
Test that model can process few - shot examples with tool calls .
"""
if not chat_model_ :
if not self . :
pytest . skip ( " Test requires tool calling. " )
model = chat_model_class ( * * chat_model_params )
model_with_tools = model . bind_tools ( [ my_adder_tool ] )
model_with_tools = model . bind_tools ( [ my_adder_tool ] , tool_choice = " any " )
function_name = " my_adder_tool "
function_args = { " a " : 1 , " b " : 2 }
function_result = json . dumps ( { " result " : 3 } )
messages_string_content = [
HumanMessage ( content = " What is 1 + 2 " ) ,
HumanMessage ( " What is 1 + 2 " ) ,
AIMessage (
content = " " ,
" " ,
tool_calls = [
{
" name " : function_name ,
@ -269,12 +203,12 @@ class ChatModelIntegrationTests(ABC):
] ,
) ,
ToolMessage (
function_result ,
name = function_name ,
content = function_result ,
tool_call_id = " abc123 " ,
) ,
AIMessage ( content= function_result) ,
HumanMessage ( content = " What is 3 + 4 " ) ,
AIMessage ( ) ,
HumanMessage ( " What is 3 + 4 " ) ,
]
result_string_content = model_with_tools . invoke ( messages_string_content )
assert isinstance ( result_string_content , AIMessage )
Bagatur
4 months ago
committed by
GitHub