File size: 2,031 Bytes
7db0ae4
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
#### What this tests ####
#    This tests calling batch_completions by running 100 messages together

import sys, os
import traceback
import pytest

sys.path.insert(
    0, os.path.abspath("../..")
)  # Adds the parent directory to the system path
from openai import APITimeoutError as Timeout
import litellm

litellm.num_retries = 0
from litellm import (
    batch_completion,
    batch_completion_models,
    completion,
    batch_completion_models_all_responses,
)

# litellm.set_verbose=True


def test_batch_completions():
    messages = [[{"role": "user", "content": "write a short poem"}] for _ in range(3)]
    model = "j2-mid"
    litellm.set_verbose = True
    try:
        result = batch_completion(
            model=model,
            messages=messages,
            max_tokens=10,
            temperature=0.2,
            request_timeout=1,
        )
        print(result)
        print(len(result))
        assert len(result) == 3
    except Timeout as e:
        print(f"IN TIMEOUT")
        pass
    except Exception as e:
        pytest.fail(f"An error occurred: {e}")


test_batch_completions()


def test_batch_completions_models():
    try:
        result = batch_completion_models(
            models=["gpt-3.5-turbo", "gpt-3.5-turbo", "gpt-3.5-turbo"],
            messages=[{"role": "user", "content": "Hey, how's it going"}],
        )
        print(result)
    except Timeout as e:
        pass
    except Exception as e:
        pytest.fail(f"An error occurred: {e}")


# test_batch_completions_models()


def test_batch_completion_models_all_responses():
    try:
        responses = batch_completion_models_all_responses(
            models=["j2-light", "claude-instant-1.2"],
            messages=[{"role": "user", "content": "write a poem"}],
            max_tokens=10,
        )
        print(responses)
        assert len(responses) == 2
    except Timeout as e:
        pass
    except Exception as e:
        pytest.fail(f"An error occurred: {e}")


# test_batch_completion_models_all_responses()