【Bug已解决】Tool use／Function calling using Langchain, AWS Bedrock and Claude does not return output 解决方案

一、现象长什么样

你用 LangChain + AWS Bedrock + Claude 做工具调用（Function Calling / Tool Use），但"没有返回输出"：

• 你用 ChatBedrock 调 bind_tools，模型返回的内容里没有 tool_use，也没触发你的函数；
• 你看到响应是空的、或只有一句"好的"却没有调用工具；
• 你用 AgentExecutor 跑，agent 直接结束，工具一次没执行；
• 你怀疑是模型不支持工具，但 Claude 明明支持；
• 你换 model_id 后有时能用有时不能；
• 你拿到的 AIMessage 里 tool_calls 为空，不知道卡在哪；
• 你不确定该用 ChatBedrock 还是 ChatBedrockConverse。

一句话：LangChain 接 Bedrock 上的 Claude 做工具调用，"没输出"通常是工具没被模型看到/没被正确解析。 常见原因：① 用的 model_id 不支持工具或格式不对；② LangChain 的 Bedrock 封装需要正确的 tools 传参与 max_tokens；③ 模型返回的是"只有 tool_use 块、没有文本"，你以为"没输出"其实是工具调用需要先执行再回传；④ 旧 ChatBedrock 走的是 Messages API 路径，工具解析和 ChatBedrockConverse（Converse API 路径）不同，选错就解析失败。

二、背景

LangChain 接 Bedrock Claude 有两条路：

1. ChatBedrock（Messages API 路径）：底层走 Bedrock 的 InvokeModel + Anthropic 原生体，工具以 Anthropic tools 格式传；
2. ChatBedrockConverse（Converse API 路径）：底层走 Bedrock 的 Converse，工具以 Converse tool 格式传，对多工具/多轮更稳，且与 LangChain tool schema 映射更干净。

"没返回输出"多发生在：工具没绑对、模型返回 tool_use 但你的代码/agent 没继续（没把 tool_result 回传）、或 max_tokens 太小导致工具调用被截断。LangChain 的 bind_tools 会把工具描述注入，但最终要靠 Agent/解析器把 tool_use 转成 tool_calls 并执行、再把结果回传——这一步断掉，就"没输出"。

三、根因

根因集中在工具绑定/解析链路断了：

# 错误示范：直接 invoke，没走 agent/executor，且没看到 tool_use
llm = ChatBedrock(model_id="anthropic.claude-3-5-sonnet-...",)
llm_with_tools = llm.bind_tools([my_tool])
msg = llm_with_tools.invoke("北京天气怎样？")
print(msg.content)   # 可能为空/只有客套话，因为模型返回的是 tool_use 而非文本
# 缺少：执行 tool -> 回传 tool_result -> 再请求

正确链路：
bind_tools -> 模型返回 tool_use(在 AIMessage.tool_calls)
-> 执行函数 -> 把 tool_result 作为 HumanMessage/tool 消息回传
-> 模型基于结果生成最终文本

若你只 invoke 一次就打印 content，且模型这次决定"先调用工具"，那你看到的 content 自然为空/客套——这不是 bug，是链路没走完。

四、最小可运行复现

用 ChatBedrockConverse + bind_tools + 手动走完工具循环的最小示例：

import boto3, json
from langchain_core.messages import HumanMessage, ToolMessage
from langchain_aws import ChatBedrockConverse

def get_weather(city: str) -> str:
    return f"{city} 今天 22℃，晴"

tools = [{
    "name": "get_weather",
    "description": "查询城市天气",
    "input_schema": {"type": "object",
        "properties": {"city": {"type": "string"}}, "required": ["city"]},
}]

llm = ChatBedrockConverse(
    model_id="anthropic.claude-3-5-sonnet-20241022-v2:0",
    region_name="us-east-1",
    max_tokens=1024,
)
llm_with_tools = llm.bind_tools([tools])

messages = [HumanMessage(content="北京天气怎样？")]
ai_msg = llm_with_tools.invoke(messages)
print("tool_calls:", ai_msg.tool_calls)   # 这里应能看到工具调用

# 走完链路：执行工具 -> 回传
if ai_msg.tool_calls:
    for call in ai_msg.tool_calls:
        res = get_weather(**call["args"])
        messages.append(ai_msg)
        messages.append(ToolMessage(content=res, tool_call_id=call["id"]))
    final = llm_with_tools.invoke(messages)
    print("最终输出:", final.content)

运行后：第一轮 tool_calls 非空（这就是"被忽略的输出"），执行并回传后第二轮才产出真正答案。

五、解决方案（第一层：最小直接修复）

最小修复是用对封装 + 走完工具循环 + 给足 max_tokens：

1. 优先用 ChatBedrockConverse（Converse 路径，工具解析更稳）；
2. 用 bind_tools 注入工具，且工具 schema 的 input_schema 必须合法；
3. 检查 ai_msg.tool_calls——非空就要执行并把 ToolMessage 回传，再 invoke 一次；
4. max_tokens 给够（工具调用结果可能较长，太小会被截断导致"看似无输出"）。

ai_msg = llm_with_tools.invoke(messages)
if ai_msg.tool_calls:                 # 关键：别只打印 content
    for call in ai_msg.tool_calls:
        result = dispatch(call["name"], call["args"])
        messages += [ai_msg, ToolMessage(content=result, tool_call_id=call["id"])]
    ai_msg = llm_with_tools.invoke(messages)   # 第二轮出最终答案
print(ai_msg.content)

要点：① "没输出"往往是只看了 content 而没处理 tool_calls；② 用 Agent/Executor 时确保工具已 bind_tools 且 executor 接管循环；③ 确认 model_id 是支持工具的 Claude 版本。

六、解决方案（第二层：结构化改进）

把"LangChain + Bedrock + Claude 工具调用"的链路固化成策略，用一个 dataclass 作为单一事实来源：

from dataclasses import dataclass, field
from typing import Callable, List


@dataclass(frozen=True)
class LangchainBedrockToolPolicy:
    """LangChain+Bedrock+Claude 工具调用策略：单一事实来源。

    规则：
    - 优先用 ChatBedrockConverse（Converse 路径，解析更稳）
    - 工具必须 bind_tools，且仅当 tool_calls 非空才执行并回传
    - max_tokens 给足，避免工具结果被截断
    - model_id 必须是支持工具的 Claude 版本
    """
    model_id: str = "anthropic.claude-3-5-sonnet-20241022-v2:0"
    region: str = "us-east-1"
    max_tokens: int = 1024

    def run(self, llm_with_tools, messages: list,
            dispatch: Callable[[str, dict], str]) -> str:
        ai = llm_with_tools.invoke(messages)
        if not getattr(ai, "tool_calls", None):
            return ai.content                 # 无需工具，直接返回
        from langchain_core.messages import ToolMessage
        for call in ai.tool_calls:
            res = dispatch(call["name"], call["args"])
            messages += [ai, ToolMessage(content=res, tool_call_id=call["id"])]
        final = llm_with_tools.invoke(messages)
        return final.content


def demo() -> None:
    policy = LangchainBedrockToolPolicy()
    assert policy.model_id.startswith("anthropic.")
    assert policy.max_tokens >= 1
    print("工具调用策略自洽")


if __name__ == "__main__":
    demo()

这样：① 工具循环统一在 run 里，不会有人只 invoke 一次就打印而漏掉 tool_calls；② Converse 路径集中，解析更稳；③ max_tokens/model 校验集中。

七、解决方案（第三层：断言 / CI 守护）

用 pytest 守住"工具链路正确处理 tool_calls"：

import pytest
from your_module import LangchainBedrockToolPolicy


def test_returns_content_when_no_tool_call():
    p = LangchainBedrockToolPolicy()
    fake_ai = type("M", (), {"tool_calls": [], "content": "直接答案"})()
    fake_llm = type("L", (), {"invoke": lambda self, m: fake_ai})()
    out = p.run(fake_llm, [], lambda n, a: "x")
    assert out == "直接答案"


def test_executes_and_returns_final():
    p = LangchainBedrockToolPolicy()
    calls = [{"name": "get_weather", "args": {"city": "北京"}, "id": "t1"}]
    first = type("M", (), {"tool_calls": calls, "content": ""})()
    final = type("M2", (), {"content": "北京 22℃ 晴"})()
    seq = {"n": 0}
    def invoke(m):
        seq["n"] += 1
        return final if seq["n"] == 2 else first
    fake_llm = type("L", (), {"invoke": invoke})()
    out = p.run(fake_llm, [], lambda n, a: "22℃")
    assert out == "北京 22℃ 晴"
    assert seq["n"] == 2     # 走了两轮


def test_model_id_is_claude():
    p = LangchainBedrockToolPolicy()
    assert p.model_id.startswith("anthropic.")


def test_max_tokens_positive():
    p = LangchainBedrockToolPolicy()
    assert p.max_tokens > 0


def test_dispatch_called_with_args():
    p = LangchainBedrockToolPolicy()
    seen = {}
    calls = [{"name": "f", "args": {"x": 1}, "id": "t"}]
    first = type("M", (), {"tool_calls": calls, "content": ""})()
    final = type("M2", (), {"content": "ok"})()
    seq = {"n": 0}
    def invoke(m):
        seq["n"] += 1
        return final if seq["n"] == 2 else first
    p.run(type("L", (), {"invoke": invoke})(), [],
          lambda n, a: seen.update(n=n, a=a) or "r")
    assert seen == {"n": "f", "a": {"x": 1}}


def test_converse_path_preferred():
    p = LangchainBedrockToolPolicy()
    # 策略默认模型即用于 ChatBedrockConverse，这里仅验证 model 形态
    assert "sonnet" in p.model_id or "claude" in p.model_id

CI 里这 6 条断言守住"无工具直接返回、有工具走两轮、dispatch 收到正确参数"，一旦有人漏掉 tool_calls 处理导致"没输出"，测试立刻红。

八、排查清单

• 是否只打印了 content 而没看 tool_calls？"没输出"常是工具调用未执行。
• 是否走了完整的"执行工具→回传 tool_result→再请求"链路？只 invoke 一次不够。
• 是否用了 ChatBedrockConverse？Converse 路径工具解析更稳。
• 工具 schema 的 input_schema 是否合法？非法会导致模型不调用。
• max_tokens 是否太小？工具结果被截断会"看似无输出"。
• model_id 是否支持工具？用 Claude 3.5/4 这类支持 tool use 的版本。
• 是否用了 AgentExecutor 却没 bind_tools？executor 需工具已绑定。
• 是否有工具链路的测试守护？没有就补一条 CI。

九、小结

LangChain + Bedrock + Claude 的"工具调用没输出"，根因多在工具绑定/解析链路断了：模型返回的是 tool_use（在 tool_calls 里），而你只看了 content。最小修复是用 ChatBedrockConverse + bind_tools，并检查 tool_calls——非空就执行函数、回传 ToolMessage 再请求一轮；结构化做法是抽成 LangchainBedrockToolPolicy 统一走完链路；最后用 pytest 守护"无工具直接返回、有工具走两轮"。这样既拿到工具结果，又不再被"空输出"误导。