如何使用 Amazon Bedrock API 实现 Function calling

### **1. 概述** 大型语言模型拥有强大的文本生成和理解能力，但在需要获取外部实时信息或执行特定功能时却存在固有的限制。为了解决这一局限，[Amazon Bedrock](https://aws.amazon.com/cn/bedrock/?trk=cndc-detail) 引入了工具调用(Tool use)功能，通常也被称为函数调用(Function calling)，本文章将使用 Bedrock中“工具调用(Tool use)”这一定义。 工具调用允许模型识别何时需要调用外部工具来完成用户请求，并指明应该使用什么参数。这使得模型能够访问实时数据、查询专有信息库或执行特定操作，从而显著提升其实用价值和应用范围。 您可以使用 [Amazon Bedrock](https://aws.amazon.com/cn/bedrock/?trk=cndc-detail) API 为模型提供工具调用的能力，使其生成更加准确、实时且有价值的响应。例如，您可能有一个聊天应用程序，允许用户查询电台播放的最受欢迎的歌曲。要响应这一请求，模型需要一个能够查询并返回歌曲信息的工具。 在这个例子中，会向模型提供一个工具定义，该工具能够返回特定电台的最受欢迎歌曲。当用户询问相关问题时，模型会分析请求并识别出需要使用该工具。但模型不会直接执行工具调用，而是返回一个调用请求，明确指出需要调用的工具及其所需参数（如电台名称），您的程序会作为模型的代理来执行工具的调用操作。您可以根据业务需求灵活实现工具——可以是 REST API、数据库查询、Lambda 函数或任何其他适合的技术实现。您的程序会将工具调用结果作为新的消息再发送给大模型。最终，模型会生成一个完整的响应，为用户提供准确答案。 您可以使用 Converse API ([Converse](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html) 或 [ConverseStream](https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ConverseStream.html))在 [Amazon Bedrock](https://aws.amazon.com/cn/bedrock/?trk=cndc-detail) 中实现工具调用的功能。本文中的示例代码将会展示如何使用 Converse API 构建一个能够查询电台热门歌曲的应用。 ### **2. 使用 Converse API 实现工具调用** 要让模型通过调用工具来完成响应，您需要向模型发送消息以及一个或多个工具的定义。如果模型确定其中某个工具可以帮助生成响应，它会返回一个请求，要求您的程序调用该工具并将结果发送回模型。随后，模型会利用这些结果为原始消息生成响应。 以下步骤展示了如何通过 Converse API 使用工具。有关示例代码，请参阅"[Converse API 工具使用示例](https://docs.aws.amazon.com/zh_cn/bedrock/latest/userguide/tool-use-examples.html)"。  ##### **2.1 向模型发送用户消息和工具定义** 工具的定义是一个JSON模式，您需要在调用 `Converse` 操作时通过 `toolConfig` 来传递参数。以下是一个工具定义示例，该工具用于获取电台播放的最受欢迎的歌曲。 ```java { "tools": [ { "toolSpec": { "name": "top_song", "description": "Get the most popular song played on a radio station.", "inputSchema": { "json": { "type": "object", "properties": { "sign": { "type": "string", "description": "The call sign for the radio station for which you want the most popular song. Example calls signs are WZPZ and WKRP." } }, "required": [ "sign" ] } } } } ] } ``` 在同一请求中，您还需要通过 `messages` 参数传递用户消息。 ```java [ { "role": "user", "content": [ { "text": "What is the most popular song on WZPZ?" } ] } ] ``` ##### **2.2 模型返回工具调用请求** 当您使用消息和工具定义调用 `Converse` API 时，模型会根据工具定义来确定是否需要使用该工具来查询信息。 例如，如果您的聊天应用用户发送消息“WZPZ 电台最受欢迎的歌曲是什么？”，模型会将这条消息和 top_song 工具定义中的描述进行匹配，并确定该工具可以帮助生成响应。 当模型决定需要使用工具来生成响应时，模型会将返回中的 `stopReason` 字段设置为 `tool_use`。模型会识别您需要使用的工具(top_song)及其所需参数（如电台编码 WZPZ），并在返回的消息中包含这些信息。您可以使用 `toolUseId` 字段在后续调用中标识工具请求。 以下示例展示了模型的响应结果。 ```java { "output": { "message": { "role": "assistant", "content": [ { "toolUse": { "toolUseId": "tooluse_kZJMlvQmRJ6eAyJE5GIl7Q", "name": "top_song", "input": { "sign": "WZPZ" } } } ] } }, "stopReason": "tool_use" } ``` ##### **2.3 应用程序代表模型执行工具请求** 应用程序会从模型响应的 `toolUse` 字段中获取 `name` 值来识别并调用您实现的工具，将工具响应结果和您在上一步中获取的工具请求 ID 作为参数传递给大模型。 ```java { "role": "user", "content": [ { "toolResult": { "toolUseId": "tooluse_kZJMlvQmRJ6eAyJE5GIl7Q", "content": [ { "json": { "song": "Elemental Hotel", "artist": "8 Storey Hike" } } ] } } ] } ``` ##### **2.4 模型返回最终响应** 接下来模型将根据您在请求参数中提供的信息，为原始问题（"WZPZ 电台最受欢迎的歌曲是什么？"）生成最终的回复。 ```java { "output": { "message": { "role": "assistant", "content": [ { "text": "The most popular song on WZPZ is Elemental Hotel by 8 Storey Hike." } ] } }, "stopReason": "end_turn" ``` ### **3. Converse API 工具使用示例** 以下 Python 示例展示了如何实现工具调用。 ```java # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. # SPDX-License-Identifier: Apache-2.0 """ Shows how to use tools with the <noloc>Converse</noloc> API and the Cohere Command R model. """ import logging import json import boto3 from botocore.exceptions import ClientError class StationNotFoundError(Exception): """Raised when a radio station isn't found.""" pass logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) def get_top_song(call_sign): """Returns the most popular song for the requested station. Args: call_sign (str): The call sign for the station for which you want the most popular song. Returns: response (json): The most popular song and artist. """ song = "" artist = "" if call_sign == 'WZPZ': song = "Elemental Hotel" artist = "8 Storey Hike" else: raise StationNotFoundError(f"Station {call_sign} not found.") return song, artist def generate_text(bedrock_client, model_id, tool_config, input_text): """Generates text using the supplied Amazon Bedrock model. If necessary, the function handles tool use requests and sends the result to the model. Args: bedrock_client: The Boto3 Bedrock runtime client. model_id (str): The Amazon Bedrock model ID. tool_config (dict): The tool configuration. input_text (str): The input text. Returns: Nothing. """ logger.info("Generating text with model %s", model_id) # Create the initial message from the user input. messages = [{ "role": "user", "content": [{"text": input_text}] }] response = bedrock_client.converse( modelId=model_id, messages=messages, toolConfig=tool_config ) output_message = response['output']['message'] messages.append(output_message) stop_reason = response['stopReason'] if stop_reason == 'tool_use': # Tool use requested. Call the tool and send the result to the model. tool_requests = response['output']['message']['content'] for tool_request in tool_requests: if 'toolUse' in tool_request: tool = tool_request['toolUse'] logger.info("Requesting tool %s. Request: %s", tool['name'], tool['toolUseId']) if tool['name'] == 'top_song': tool_result = {} try: song, artist = get_top_song(tool['input']['sign']) tool_result = { "toolUseId": tool['toolUseId'], "content": [{"json": {"song": song, "artist": artist}}] } except StationNotFoundError as err: tool_result = { "toolUseId": tool['toolUseId'], "content": [{"text": err.args[0]}], "status": 'error' } tool_result_message = { "role": "user", "content": [ { "toolResult": tool_result } ] } messages.append(tool_result_message) # Send the tool result to the model. response = bedrock_client.converse( modelId=model_id, messages=messages, toolConfig=tool_config ) output_message = response['output']['message'] # print the final response from the model. for content in output_message['content']: print(json.dumps(content, indent=4)) def main(): """ Entrypoint for tool use example. """ logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s") model_id = "cohere.command-r-v1:0" input_text = "What is the most popular song on WZPZ?" tool_config = { "tools": [ { "toolSpec": { "name": "top_song", "description": "Get the most popular song played on a radio station.", "inputSchema": { "json": { "type": "object", "properties": { "sign": { "type": "string", "description": "The call sign for the radio station for which you want the most popular song. Example calls signs are WZPZ, and WKRP." } }, "required": [ "sign" ] } } } } ] } bedrock_client = boto3.client(service_name='bedrock-runtime') try: print(f"Question: {input_text}") generate_text(bedrock_client, model_id, tool_config, input_text) except ClientError as err: message = err.response['Error']['Message'] logger.error("A client error occurred: %s", message) print(f"A client error occured: {message}") else: print( f"Finished generating text with model {model_id}.") if __name__ == "__main__": main() ``` 有关其他代码示例，请参阅"[使用 Amazon SDK 的 Amazon Bedrock Runtime 代码示例](https://docs.aws.amazon.com/zh_cn/bedrock/latest/userguide/service_code_examples_bedrock-runtime.html)"。