Bedrock上的Claude模型的Tool use

一、背景

1、什么是Tool use

Tool use也叫做Function calling，这是指模型识别访问意图并调用外部工具的能力。例如在一个对话查询中，希望检索互联网上当前最火热的歌曲，或者触发另一个系统的特定的API。这种能力往往和Agent以及知识库搭配使用。需要注意的是，Tool use场景中大语言模型不会直接运行API Call，而是将需要API Call的请求拼接好返回给调用大语言模型的代码。API call的执行过程是完全由程序调用来负责执行的。因此当代码执行API Call获得返回结果之后，还需要将返回结果再次输入到大语言模型中，并且包含上次的聊天记录一起返回。这时即可获得预期的插叙结果。

本文以一个数学计算为例，输入一个计算要求，识别是Tool场景，程序完成Tool use获取结果，再将结果代回到大模型对话，完成整个流程。

二、识别Tool use场景

事先安装Python的最新版的AWS SDK的依赖库文件，如果是旧版本，升级到最新版Boto3。

pip3 install boto3 --upgrade

编写如下Python代码，替换其中的Region和模型名字为实际的环境以及要调用的模型的名字。在以下代码中，通过System Prompt指定了数学计算要通过Tool use来进行。由此识别出来Tool use，然后构建出来JSON文件用于Tool use调用。

import boto3, json, math

region_name = 'us-west-2'
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

session = boto3.Session(region_name= region_name)
bedrock = session.client(service_name="bedrock-runtime")

tool_list = [
    {
        "toolSpec": {
            "name": "cosine",
            "description": "Calculate the cosine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    }
]

message_list = []

initial_message = {
    "role": "user",
    "content": [
        { "text": "What is the cosine of 7?" } 
    ],
}

message_list.append(initial_message)

response = bedrock.converse(
    modelId=model_id,
    messages=message_list,
    inferenceConfig={
        "maxTokens": 2000,
        "temperature": 0
    },
    toolConfig={
        "tools": tool_list
    },
    system=[{"text":"You must only do math by using a tool."}]
)

response_message = response['output']['message']
message_list.append(response_message)

# print message for debug
print(json.dumps(response_message, indent=4))

返回信息如下：

{
    "role": "assistant",
    "content": [
        {
            "text": "Here is how we can calculate the cosine of 7 using the available tool:"
        },
        {
            "toolUse": {
                "toolUseId": "tooluse_iee0b75bTnagdKYmgstAHw",
                "name": "cosine",
                "input": {
                    "x": 7
                }
            }
        }
    ]
}

这表示模型识别出来了这是一个需要Tool use调用的场景，并且将返回结果构建为统一的JSON形式，方便后续代码编写。

三、执行Tool并直接返回结果

在上一步的基础上更进一步，模型判断出来了是调用Tool use，那么就在代码中直接计算，并生成结果。注意，这里的计算结果不返回给模型，而是直接从Python程序一侧调用Tool算好了就print出来。代码如下。

import boto3, json, math

region_name = 'us-west-2'
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

session = boto3.Session(region_name= region_name)
bedrock = session.client(service_name="bedrock-runtime")

tool_list = [
    {
        "toolSpec": {
            "name": "cosine",
            "description": "Calculate the cosine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    }
]

message_list = []

initial_message = {
    "role": "user",
    "content": [
        { "text": "What is the cosine of 7?" } 
    ],
}

message_list.append(initial_message)

response = bedrock.converse(
    modelId=model_id,
    messages=message_list,
    inferenceConfig={
        "maxTokens": 2000,
        "temperature": 0
    },
    toolConfig={
        "tools": tool_list
    },
    system=[{"text":"You must only do math by using a tool."}]
)

response_message = response['output']['message']
message_list.append(response_message)

response_content_blocks = response_message['content']

for content_block in response_content_blocks:
    if 'toolUse' in content_block:
        tool_use_block = content_block['toolUse']
        tool_use_name = tool_use_block['name']
        
        print(f"Using tool {tool_use_name}")
        
        if tool_use_name == 'cosine':
            tool_result_value = math.cos(math.radians(tool_use_block['input']['x']))
            print(tool_result_value)
            
    elif 'text' in content_block:
        print(content_block['text'])

返回结果如下：

Here is how we can calculate the cosine of 7 using the available tool:
Using tool cosine
0.992546151641322

可以看到模型返回说要调用Tool之后，就靠Tool自己完成任务了。计算结果不会代入回模型，因此Bedrock服务中模型一侧只接受到了一次调用，模型是不知道计算结果的。

如果希望将计算结果返回给模型进一步处理，那么可以使用如下代码。

四、将Tool运行结果代入模型交互

在上一代断码的最后一个content_block in response_content_blocks循环中，可将Tool给出的结果，包含确认要使用Tool的历史对话记录，再次代入到模型中，可让模型根据返回结果再做进一步处理。代码如下。

import boto3, json, math

region_name = 'us-west-2'
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

session = boto3.Session(region_name= region_name)
bedrock = session.client(service_name="bedrock-runtime")

tool_list = [
    {
        "toolSpec": {
            "name": "cosine",
            "description": "Calculate the cosine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    }
]

message_list = []

initial_message = {
    "role": "user",
    "content": [
        { "text": "What is the cosine of 7?" } 
    ],
}

message_list.append(initial_message)

response = bedrock.converse(
    modelId=model_id,
    messages=message_list,
    inferenceConfig={
        "maxTokens": 2000,
        "temperature": 0
    },
    toolConfig={
        "tools": tool_list
    },
    system=[{"text":"You must only do math by using a tool."}]
)

response_message = response['output']['message']
message_list.append(response_message)

response_content_blocks = response_message['content']

follow_up_content_blocks = []

for content_block in response_content_blocks:
    if 'toolUse' in content_block:
        tool_use_block = content_block['toolUse']
        tool_use_name = tool_use_block['name']
        
        
        if tool_use_name == 'cosine':
            tool_result_value = math.cos(math.radians(tool_use_block['input']['x']))
            
            follow_up_content_blocks.append({
                "toolResult": {
                    "toolUseId": tool_use_block['toolUseId'],
                    "content": [
                        {
                            "json": {
                                "result": tool_result_value
                            }
                        }
                    ]
                }
            })

if len(follow_up_content_blocks) > 0:
    
    follow_up_message = {
        "role": "user",
        "content": follow_up_content_blocks,
    }
    
    message_list.append(follow_up_message)

    response = bedrock.converse(
        modelId=model_id,
        messages=message_list,
        inferenceConfig={
            "maxTokens": 2000,
            "temperature": 0
        },
        toolConfig={
            "tools": tool_list
        },
        system=[{"text":"You must only do math by using a tool."}]
    )
    
    response_message = response['output']['message']
    print(json.dumps(response_message, indent=4))
    message_list.append(response_message)

返回结果：

{
    "role": "assistant",
    "content": [
        {
            "text": "So the cosine of 7 is 0.7539022543433046."
        }
    ]
}

由此可看到模型处理了Tool use返回的结果。这样加起来两次模型调用，中间一次Tool use，就完成了整个流程。

五、错误处理

在上边的演示中，Tool use的过程没有使用真实的第三方API调用，而是直接用Python的math数学库进行了计算，模拟了一次Tool use。在实际使用环境中，Tool use调用的第三方API可能是存在网络可达、连接失败、超时、没有正常返回预期结果、QoS限流等多种技术问题，此时Tool use调用结果是一堆错误信息，是千奇百怪的。这些错误信息如果返回给大语言模型，刚才的对话和Tool use是无法给出正确答案的，此时需要错误处理。

为了解决Tool use调用第三方API可能的失效问题，可在Tool use返回JSON返回中增加一个标签"status": "error"表示Tool use的返回结果不可用。这样模型就会理解为此时Tool use调用的第三方API不可用，并且模型最后会给出友好的回答，而不是生硬的抛出错误，或者胡乱回答。这里可通过再次调用大模型时候在Prompt中添加Tool use不可用时候的提示词来解决。

代码样例如下。

import boto3, json, math

region_name = 'us-west-2'
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

session = boto3.Session(region_name= region_name)
bedrock = session.client(service_name="bedrock-runtime")

tool_list = [
    {
        "toolSpec": {
            "name": "cosine",
            "description": "Calculate the cosine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    }
]

message_list = []

initial_message = {
    "role": "user",
    "content": [
        { "text": "What is the cosine of 7?" } 
    ],
}

message_list.append(initial_message)

response = bedrock.converse(
    modelId=model_id,
    messages=message_list,
    inferenceConfig={
        "maxTokens": 2000,
        "temperature": 0
    },
    toolConfig={
        "tools": tool_list
    },
    system=[{"text":"You must only do math by using a tool."}]
)

response_message = response['output']['message']
message_list.append(response_message)

response_content_blocks = response_message['content']

follow_up_content_blocks = []

for content_block in response_content_blocks:
    if 'toolUse' in content_block:
        tool_use_block = content_block['toolUse']
        tool_use_name = tool_use_block['name']
        
        
        if tool_use_name == 'cosine':
            tool_result_value = math.cos(tool_use_block['input']['x'])
            
            follow_up_content_blocks.append({
                "toolResult": {
                    "toolUseId": tool_use_block['toolUseId'],
                    "content": [
                        {
                            "text": "invalid function: cosine"
                        }
                    ],
                "status": "error"
                }
            })

if len(follow_up_content_blocks) > 0:
    
    follow_up_message = {
        "role": "user",
        "content": follow_up_content_blocks,
    }
    
    message_list.append(follow_up_message)

    response = bedrock.converse(
        modelId=model_id,
        messages=message_list,
        inferenceConfig={
            "maxTokens": 2000,
            "temperature": 0
        },
        toolConfig={
            "tools": tool_list
        },
        system=[{"text":"You must only do math by using a tool. If tool is not available, do not assume, do not provide more info, just say the tool is unavailable."}]
    )
    
    response_message = response['output']['message']
    print(json.dumps(response_message, indent=4))
    message_list.append(response_message)

返回结果如下：

{
    "role": "assistant",
    "content": [
        {
            "text": "Unfortunately, the \"cosine\" tool is not available in this environment. Without access to that mathematical function, I cannot provide the cosine of 7. Please let me know if there are any other tools available that I could use to assist with your request."
        }
    ]
}

由此可以看到，Tool use不可用时候，模型也返回了友好信息。

六、多个Tool的识别和匹配场景

如果在一个业务场景中，需要处理多个Tool调用的场景，那么可以定义两个Tool在名称上区分开，然后不同的提问匹配到不通的Tool调用，即可完成业务的选择。代码如下。

import boto3, json, math

region_name = 'us-west-2'
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

session = boto3.Session(region_name= region_name)
bedrock = session.client(service_name="bedrock-runtime")

tool_list = [
    {
        "toolSpec": {
            "name": "cosine",
            "description": "Calculate the cosine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    },
        {
        "toolSpec": {
            "name": "sine",
            "description": "Calculate the sine of x.",
            "inputSchema": {
                "json": {
                    "type": "object",
                    "properties": {
                        "x": {
                            "type": "number",
                            "description": "The number to pass to the function."
                        }
                    },
                    "required": ["x"]
                }
            }
        }
    }
]

message_list = []

initial_message = {
    "role": "user",
    "content": [
        { "text": "What is the sine of 7?" } 
    ],
}

message_list.append(initial_message)

response = bedrock.converse(
    modelId=model_id,
    messages=message_list,
    inferenceConfig={
        "maxTokens": 2000,
        "temperature": 0
    },
    toolConfig={
        "tools": tool_list
    },
    system=[{"text":"You must only do math by using a tool."}]
)

response_message = response['output']['message']
message_list.append(response_message)

# print message for debug
print(json.dumps(response_message, indent=4))

response_content_blocks = response_message['content']

follow_up_content_blocks = []

for content_block in response_content_blocks:
    if 'toolUse' in content_block:
        tool_use_block = content_block['toolUse']
        tool_use_name = tool_use_block['name']
        
        if tool_use_name == 'cosine':
            tool_result_value = math.cos(math.radians(tool_use_block['input']['x']))

        if tool_use_name == 'sine':
            tool_result_value = math.sin(math.radians(tool_use_block['input']['x']))
            
        # debug
        print(tool_result_value)
            
        follow_up_content_blocks.append({
            "toolResult": {
                "toolUseId": tool_use_block['toolUseId'],
                "content": [
                        {
                            "json": {
                                "result": tool_result_value
                            }
                        }
                    ]
            }
        })

if len(follow_up_content_blocks) > 0:
    
    follow_up_message = {
        "role": "user",
        "content": follow_up_content_blocks,
    }
    
    message_list.append(follow_up_message)

    response = bedrock.converse(
        modelId=model_id,
        messages=message_list,
        inferenceConfig={
            "maxTokens": 2000,
            "temperature": 0
        },
        toolConfig={
            "tools": tool_list
        },
        system=[{"text":"You must only do math by using a tool."}]
    )
    
    response_message = response['output']['message']
    print(json.dumps(response_message, indent=4))
    message_list.append(response_message)

修改以上代码中的提问，分别询问计算sin和cos函数的要求，即可看到正确使用了tool并返回结果。

七、参考文档

Intro to Tool Use with the Amazon Bedrock Converse API

https://community.aws/content/2hW5367isgQOkkXLYjp4JB3Pe16/intro-to-tool-use-with-the-amazon-bedrock-converse-api

Call a tool with Amazon Bedrock Tool use (Function calling)

https://docs.aws.amazon.com/bedrock/latest/userguide/tool-use.html