Skip to content

Anthropic

Generates texts with using the Anthropic Batch API.

AnthropicClient

Bases: BatchClient

Source code in dactyl_generation/anthropic_generation.py 
 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
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class AnthropicClient(BatchClient):

    def __init__(self, api_key: str, version: str) -> None:
        """_summary_
        Constructor for Anthropic client. 

        Args:
            api_key: Anthropic API key.
            version: Anthropic API version to use. 
        """
        super().__init__()
        self.api_key = api_key
        self.client = anthropic.Anthropic(
            api_key=api_key,
        )
        self.api_headers = {"x-api-key": self.api_key, "anthropic-version": version}



    @staticmethod
    def convert_openai_system_message_to_anthropic_system_message(openai_message: dict) -> dict:
        """
        Converts OpenAI system message to Anthropic API system message.
        Doesn't support cache control yet!

        Args:
            openai_message: dictionary containing system prompt

        Returns:
            anthropic_system_prompt: dictionary containing Anthropic API message
        """
        ret = dict()
        ret[TEXT] = openai_message[CONTENT]
        ret[TYPE] = TEXT
        return ret

    @staticmethod
    def convert_anthropic_system_message_to_openai_system_message(anthropic_message: dict) -> dict:
        """
        Converts Anthropic API system message to OpenAI API system message.
        Doesn't support cache control yet!

        Args:
            anthropic_message: dictionary containing system prompt

        Returns:
            anthropic_system_prompt: dictionary containing Anthropic API message
        """
        ret = dict()
        ret[ROLE] = SYSTEM
        ret[CONTENT] = anthropic_message[TEXT]
        return ret

    @staticmethod
    def get_message_batch(prompts_df: pd.DataFrame) -> List[Request]:
        """
        Generate a batch of requests from list of prompts

        Args:
            prompts_df: DataFrame where each row is an API call to the Anthropic API.

        Returns:
            requests: list of requests
        """
        requests = list()
        calls = prompts_df.drop(columns=[CUSTOM_ID]).to_dict(orient="records")
        for i, call in enumerate(calls):
            system_messages = list()
            normal_messages = list()
            for message in call[PROMPT]:
                if message[ROLE] == SYSTEM:
                    system_messages.append(AnthropicClient.convert_openai_system_message_to_anthropic_system_message(message))
                else:
                    normal_messages.append(message)

            call[SYSTEM] = system_messages
            call[MESSAGES] = normal_messages
            message_parameters = copy.copy(call)
            del message_parameters[PROMPT]
            # each individual request maps to one few shot set
            request = Request(
                custom_id=prompts_df[CUSTOM_ID].values[i],
                params=MessageCreateParamsNonStreaming(
                    **message_parameters
                )
            )
            requests.append(request)
        return requests


    def create_batch_job(self, prompts_df: pd.DataFrame) -> dict:
        """
        Requests message batch to Anthropic API given a list of examples.

        Args:
            prompts_df: Dataframe containing prompts to run.

        Returns:
            request_data: requests sent to Anthropic API
        """


        requests = AnthropicClient.get_message_batch(prompts_df)
        custom_ids = [request[CUSTOM_ID] for request in requests]
        message_batch = self.client.messages.batches.create(requests=requests)
        prompts_df[CUSTOM_ID] = custom_ids
        return {
            BATCH_ID: message_batch.id,
            PROMPTS: prompts_df.to_dict(orient='records'),
            API_CALL: ANTHROPIC,
            TIMESTAMP:  str(datetime.now(timezone.utc))
        }



    def get_batch_job_output(self, file_path: str) -> pd.DataFrame:
        """
        Gets batch job results using saved metadata from a local JSON file.

        Args:
            file_path: local JSON file containing output of the `request_batch_job` function

        Returns:
            df: pandas DataFrame of generations.
        """
        with open(file_path) as f:
            data = json.load(f)
        message_id = data[BATCH_ID]
        response = requests.get(f"https://api.anthropic.com/v1/messages/batches/{message_id}/results",headers=self.api_headers)
        lines = response.text.splitlines()
        objects = list()
        for line in lines:
            objects.append(json.loads(line))
        generations = list()
        for object in objects:
            generation = dict()
            generation[CUSTOM_ID] = object[CUSTOM_ID]
            generation[TEXT] = object[RESULT][MESSAGE][CONTENT][0][TEXT]
            generations.append(generation)
        generations = pd.DataFrame(generations)
        generations[TIMESTAMP] = data[TIMESTAMP]
        prompt_rows = pd.DataFrame(data[PROMPTS])
        ret = pd.DataFrame(prompt_rows)
        return generations.merge(ret, on=CUSTOM_ID, how='left')

__init__(api_key, version)

summary Constructor for Anthropic client.

Parameters:

Name Type Description Default
api_key str

Anthropic API key.

 required
version str

Anthropic API version to use.

 required
Source code in dactyl_generation/anthropic_generation.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
def __init__(self, api_key: str, version: str) -> None:
    """_summary_
    Constructor for Anthropic client. 

    Args:
        api_key: Anthropic API key.
        version: Anthropic API version to use. 
    """
    super().__init__()
    self.api_key = api_key
    self.client = anthropic.Anthropic(
        api_key=api_key,
    )
    self.api_headers = {"x-api-key": self.api_key, "anthropic-version": version}

convert_anthropic_system_message_to_openai_system_message(anthropic_message) staticmethod

Converts Anthropic API system message to OpenAI API system message. Doesn't support cache control yet!

Parameters:

Name Type Description Default
anthropic_message dict

dictionary containing system prompt

 required

Returns:

Name Type Description
anthropic_system_prompt dict

dictionary containing Anthropic API message
Source code in dactyl_generation/anthropic_generation.py 
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
@staticmethod
def convert_anthropic_system_message_to_openai_system_message(anthropic_message: dict) -> dict:
    """
    Converts Anthropic API system message to OpenAI API system message.
    Doesn't support cache control yet!

    Args:
        anthropic_message: dictionary containing system prompt

    Returns:
        anthropic_system_prompt: dictionary containing Anthropic API message
    """
    ret = dict()
    ret[ROLE] = SYSTEM
    ret[CONTENT] = anthropic_message[TEXT]
    return ret

convert_openai_system_message_to_anthropic_system_message(openai_message) staticmethod

Converts OpenAI system message to Anthropic API system message. Doesn't support cache control yet!

Parameters:

Name Type Description Default
openai_message dict

dictionary containing system prompt

 required

Returns:

Name Type Description
anthropic_system_prompt dict

dictionary containing Anthropic API message
Source code in dactyl_generation/anthropic_generation.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
@staticmethod
def convert_openai_system_message_to_anthropic_system_message(openai_message: dict) -> dict:
    """
    Converts OpenAI system message to Anthropic API system message.
    Doesn't support cache control yet!

    Args:
        openai_message: dictionary containing system prompt

    Returns:
        anthropic_system_prompt: dictionary containing Anthropic API message
    """
    ret = dict()
    ret[TEXT] = openai_message[CONTENT]
    ret[TYPE] = TEXT
    return ret

create_batch_job(prompts_df)

Requests message batch to Anthropic API given a list of examples.

Parameters:

Name Type Description Default
prompts_df DataFrame

Dataframe containing prompts to run.

 required

Returns:

Name Type Description
request_data dict

requests sent to Anthropic API
Source code in dactyl_generation/anthropic_generation.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def create_batch_job(self, prompts_df: pd.DataFrame) -> dict:
    """
    Requests message batch to Anthropic API given a list of examples.

    Args:
        prompts_df: Dataframe containing prompts to run.

    Returns:
        request_data: requests sent to Anthropic API
    """


    requests = AnthropicClient.get_message_batch(prompts_df)
    custom_ids = [request[CUSTOM_ID] for request in requests]
    message_batch = self.client.messages.batches.create(requests=requests)
    prompts_df[CUSTOM_ID] = custom_ids
    return {
        BATCH_ID: message_batch.id,
        PROMPTS: prompts_df.to_dict(orient='records'),
        API_CALL: ANTHROPIC,
        TIMESTAMP:  str(datetime.now(timezone.utc))
    }

get_batch_job_output(file_path)

Gets batch job results using saved metadata from a local JSON file.

Parameters:

Name Type Description Default
file_path str

local JSON file containing output of the request_batch_job function

 required

Returns:

Name Type Description
df DataFrame

pandas DataFrame of generations.
Source code in dactyl_generation/anthropic_generation.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def get_batch_job_output(self, file_path: str) -> pd.DataFrame:
    """
    Gets batch job results using saved metadata from a local JSON file.

    Args:
        file_path: local JSON file containing output of the `request_batch_job` function

    Returns:
        df: pandas DataFrame of generations.
    """
    with open(file_path) as f:
        data = json.load(f)
    message_id = data[BATCH_ID]
    response = requests.get(f"https://api.anthropic.com/v1/messages/batches/{message_id}/results",headers=self.api_headers)
    lines = response.text.splitlines()
    objects = list()
    for line in lines:
        objects.append(json.loads(line))
    generations = list()
    for object in objects:
        generation = dict()
        generation[CUSTOM_ID] = object[CUSTOM_ID]
        generation[TEXT] = object[RESULT][MESSAGE][CONTENT][0][TEXT]
        generations.append(generation)
    generations = pd.DataFrame(generations)
    generations[TIMESTAMP] = data[TIMESTAMP]
    prompt_rows = pd.DataFrame(data[PROMPTS])
    ret = pd.DataFrame(prompt_rows)
    return generations.merge(ret, on=CUSTOM_ID, how='left')

get_message_batch(prompts_df) staticmethod

Generate a batch of requests from list of prompts

Parameters:

Name Type Description Default
prompts_df DataFrame

DataFrame where each row is an API call to the Anthropic API.

 required

Returns:

Name Type Description
requests List[Request]

list of requests
Source code in dactyl_generation/anthropic_generation.py 
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
@staticmethod
def get_message_batch(prompts_df: pd.DataFrame) -> List[Request]:
    """
    Generate a batch of requests from list of prompts

    Args:
        prompts_df: DataFrame where each row is an API call to the Anthropic API.

    Returns:
        requests: list of requests
    """
    requests = list()
    calls = prompts_df.drop(columns=[CUSTOM_ID]).to_dict(orient="records")
    for i, call in enumerate(calls):
        system_messages = list()
        normal_messages = list()
        for message in call[PROMPT]:
            if message[ROLE] == SYSTEM:
                system_messages.append(AnthropicClient.convert_openai_system_message_to_anthropic_system_message(message))
            else:
                normal_messages.append(message)

        call[SYSTEM] = system_messages
        call[MESSAGES] = normal_messages
        message_parameters = copy.copy(call)
        del message_parameters[PROMPT]
        # each individual request maps to one few shot set
        request = Request(
            custom_id=prompts_df[CUSTOM_ID].values[i],
            params=MessageCreateParamsNonStreaming(
                **message_parameters
            )
        )
        requests.append(request)
    return requests