Langchain+AWS Bedrock+Claude调用自定义工具无输出问题排查

• 查询天气时，模型正确识别并返回tool_use指令，但未触发工具执行，仅输出token计数，无工具返回结果
• 无需工具的常规查询（如问印度首都）可正常返回答案

原代码实现 #

import os
from dotenv import find_dotenv, load_dotenv
dotenv_path = find_dotenv()
load_dotenv(dotenv_path)

import json
import boto3
from langchain_community.llms import Bedrock

aws_region = "us-east-1"
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

model = boto3.client(service_name="bedrock-runtime", region_name="us-east-1")

# 自定义工具函数
def get_weather(location: str) -> str:
    return f"The weather in {location} is sunny and warm."

# 工具Schema定义
tool_schema = {
    "type": "object",
    "properties": {
        "location": {"type": "string", "description": "The city and state, e.g. San Francisco, CA"}
    },
    "required": ["location"]
}

# 工具定义
tool = [{
    "name": "get_weather",
    "description": "Get the current weather in a given location",
    "input_schema": tool_schema
}]

# 冗余工具定义（未使用）
tools = [
    {"name": "get_weather", "description": "Get the current weather in a given location"},
]

def query(prompt):
    body = json.dumps({
        "max_tokens": 256,
        "messages": [{"role": "user", "content": prompt}],
        "anthropic_version": "bedrock-2023-05-31",
        "tools": tool
    })

    response = model.invoke_model(body=body, modelId=model_id)
    streaming_body = response["body"]
    s = streaming_body.read().decode('utf-8')
    sJson = json.loads(s)
    
    print("------------------------------------")
    print("q: " + prompt)
    print(json.dumps(sJson, indent=4))
    
query("What is the weather like in San Francisco?")
query("What is the capital of India?")

原运行输出 #

------------------------------------
q: What is the weather like in San Francisco?
{
    "id": "msg_bdrk_01KCn2u3P2f2WCiQ4yuvkpNL",
    "type": "message",
    "role": "assistant",
    "model": "claude-3-sonnet-20240229",
    "content": [
        {
            "type": "text",
            "text": "Okay, let's get the current weather for San Francisco:"
        },
        {
            "type": "tool_use",
            "id": "toolu_bdrk_01J6xY1kfNA1dVMbkqCpAp3K",
            "name": "get_weather",
            "input": {
                "location": "San Francisco, CA"
            }
        }
    ],
    "stop_reason": "tool_use",
    "stop_sequence": null,
    "usage": {
        "input_tokens": 249,
        "output_tokens": 70
    }
}
------------------------------------
q: What is the capital of India?
{
    "id": "msg_bdrk_011qdVBM8cAq69Z7w7TnVsMs",
    "type": "message",
    "role": "assistant",
    "model": "claude-3-sonnet-20240229",
    "content": [
        {
            "type": "text",
            "text": "The capital of India is New Delhi."
        }
    ],
    "stop_reason": "end_turn",
    "stop_sequence": null,
    "usage": {
        "input_tokens": 248,
        "output_tokens": 11
    }
}

问题原因与解决方案 #

核心问题 #

代码仅完成了工具定义发送和模型初始响应接收，缺少以下关键环节：

1. 解析模型返回的tool_use指令
2. 调用对应的自定义工具函数
3. 将工具执行结果回传给模型，生成最终自然语言回答

修正后的完整代码 #

import os
from dotenv import find_dotenv, load_dotenv
dotenv_path = find_dotenv()
load_dotenv(dotenv_path)

import json
import boto3

aws_region = "us-east-1"
model_id = "anthropic.claude-3-sonnet-20240229-v1:0"

model = boto3.client(service_name="bedrock-runtime", region_name="us-east-1")

# 自定义工具函数
def get_weather(location: str) -> str:
    return f"The weather in {location} is sunny and warm."

# 工具Schema定义
tool_schema = {
    "type": "object",
    "properties": {
        "location": {"type": "string", "description": "The city and state, e.g. San Francisco, CA"}
    },
    "required": ["location"]
}

# 工具定义
tools = [{
    "name": "get_weather",
    "description": "Get the current weather in a given location",
    "input_schema": tool_schema
}]

# 工具映射：用于根据工具名调用对应函数
tool_map = {
    "get_weather": get_weather
}

def handle_tool_use(tool_use):
    """解析工具调用请求并执行对应工具"""
    tool_name = tool_use["name"]
    tool_input = tool_use["input"]
    tool_func = tool_map.get(tool_name)
    
    if not tool_func:
        return f"Error: Tool {tool_name} not found"
    
    # 根据工具参数执行函数
    if tool_name == "get_weather":
        return tool_func(tool_input["location"])
    return ""

def query(prompt):
    messages = [{"role": "user", "content": prompt}]
    
    while True:
        body = json.dumps({
            "max_tokens": 256,
            "messages": messages,
            "anthropic_version": "bedrock-2023-05-31",
            "tools": tools
        })

        response = model.invoke_model(body=body, modelId=model_id)
        streaming_body = response["body"]
        s = streaming_body.read().decode('utf-8')
        sJson = json.loads(s)
        
        print("------------------------------------")
        print("q: " + prompt)
        print(json.dumps(sJson, indent=4))
        
        # 检查是否需要调用工具
        if sJson["stop_reason"] == "tool_use":
            # 提取工具调用指令
            tool_use = next(item for item in sJson["content"] if item["type"] == "tool_use")
            # 执行工具
            tool_result = handle_tool_use(tool_use)
            # 将工具结果添加到消息列表，发起第二轮请求
            messages.extend([
                sJson,  # 模型的tool_use响应
                {
                    "role": "user",
                    "content": [
                        {
                            "type": "tool_result",
                            "tool_use_id": tool_use["id"],
                            "content": tool_result
                        }
                    ]
                }
            ])
        else:
            # 无需工具，结束循环
            break

query("What is the weather like in San Francisco?")
query("What is the capital of India?")

关键修改说明 #

1. 新增工具映射与执行逻辑：通过tool_map关联工具名和函数，handle_tool_use负责解析模型的工具调用请求并执行对应函数
2. 多轮对话处理：在query函数中加入循环，当模型返回stop_reason="tool_use"时，将工具结果以tool_result格式回传给模型，触发第二轮生成
3. 清理冗余定义：删除未使用的tools冗余列表，统一工具定义规范

内容的提问来源于stack exchange，提问作者Sujata Pradhan