Claude 中转流式输出与 Function Calling 实战

KingFlow · 国内直连 AI API 中转

写这篇之前我踩过不少坑：把中转 Base URL 一换，普通对话跑得好好的，一开 stream=True 就卡住不吐字，或者一上 tools 参数直接 400。后来才明白，中转站能不能"官方协议完整透传"，流式和工具调用是两块最容易露馅的试金石。这篇就用能直接跑的代码，把这两件事讲透。

一、流式为什么重要，中转要能透传 SSE

先说流式（streaming）到底解决什么问题。

你让模型写一段两三百字的回复，非流式模式下，客户端要等模型把整段生成完、服务端一次性返回，用户盯着转圈可能得等好几秒甚至十几秒。而流式是模型每生成几个 token 就往回推一次，前端能像打字机一样逐字显示。对用户体验的差别是决定性的：

首字延迟（TTFT）体感大幅下降——用户看到第一个字开始跳出来，就觉得"它在干活了",焦虑感立刻消失；
长回复不再是黑盒——生成一大段代码或文章时，能边出边看，不用赌最后一口气；
可随时打断——发现方向不对，中途就能停，省 token。

技术上，Claude 的流式走的是 SSE（Server-Sent Events），Content-Type: text/event-stream，服务端一条条 event: / data: 往下推，连接保持不关。

关键点来了：中转站必须"透传"这个 SSE 流，而不是在自己后端把整段收完再一次性返回。 有些实现图省事，内部等 Claude 全部生成完再吐给你，客户端设了 stream=True 却毫无打字机效果——这就是假流式。判断方法很简单：开流式发一个长回复请求，掐秒表看第一个 chunk 什么时候到。如果和非流式差不多时间才开始出字，那它没真透传。KingFlow 这类走官方 /v1/messages 协议的中转，SSE 是原样透传的，首字通常一两秒就开始跳。

二、流式代码示例（Python）

直接上能跑的。用官方 anthropic SDK，只改 base_url 指向中转端点即可：

from anthropic import Anthropic

client = Anthropic(
    api_key="你的_KingFlow_Key",
    base_url="https://www.kingflow.ai/v1",
)

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[
        {"role": "user", "content": "用三段话讲讲流式输出的原理"}
    ],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
    print()

    # 流结束后可以拿到完整的最终消息对象
    final = stream.get_final_message()
    print("\n用量：", final.usage)

stream.text_stream 是 SDK 帮你封装好的文本增量迭代器，end="", flush=True 保证逐字打到终端上。

如果你想自己处理原始事件（比如做前端 SSE 转发），可以遍历事件流，按类型分发：

with client.messages.stream(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "你好"}],
) as stream:
    for event in stream:
        if event.type == "content_block_delta":
            # 文本增量在这里
            if event.delta.type == "text_delta":
                print(event.delta.text, end="", flush=True)
        elif event.type == "message_stop":
            print("\n[结束]")

这里几个事件类型值得记：message_start（消息开始，带初始 usage）、content_block_delta（内容增量，文本或工具参数都走这个）、message_delta（携带 stop_reason 等）、message_stop（收尾）。做过一次原始事件处理，你对后面工具调用的流式就不会陌生。

三、Function Calling / 工具调用代码示例

工具调用（Function Calling）是让模型不只是聊天，而是能"决定调用你提供的函数"。你在请求里用 tools 声明有哪些工具、每个工具的入参 schema，模型判断需要时就返回一个 tool_use 块，把参数填好交给你，你执行完再把结果喂回去。

一个完整的天气查询例子：

from anthropic import Anthropic

client = Anthropic(
    api_key="你的_KingFlow_Key",
    base_url="https://www.kingflow.ai/v1",
)

tools = [
    {
        "name": "get_weather",
        "description": "查询指定城市的实时天气",
        "input_schema": {
            "type": "object",
            "properties": {
                "city": {"type": "string", "description": "城市名，如 北京"}
            },
            "required": ["city"],
        },
    }
]

messages = [{"role": "user", "content": "上海现在天气怎么样？"}]

resp = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=messages,
)

# 模型决定调用工具时，stop_reason 会是 "tool_use"
if resp.stop_reason == "tool_use":
    # 先把模型这一轮（含 tool_use 块）追加进对话历史
    messages.append({"role": "assistant", "content": resp.content})

    for block in resp.content:
        if block.type == "tool_use":
            print("模型要调用：", block.name, block.input)
            # 这里换成你真正的函数
            result = f"{block.input['city']}：晴，28℃"
            # 把工具结果作为 tool_result 回传，注意 tool_use_id 要对上
            messages.append({
                "role": "user",
                "content": [{
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result,
                }],
            })

    # 再请求一次，模型拿到结果后生成自然语言回复
    final = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        tools=tools,
        messages=messages,
    )
    print(final.content[0].text)

几个容易翻车的细节：

tool_use_id 必须原样回传——tool_result 里的 id 要和模型返回的 tool_use 块 id 完全一致，对不上模型就不认。
assistant 这一轮要完整追加——包含 tool_use 块的整条 assistant 消息要进历史，不能只塞工具结果。
工具调用也能流式——工具的参数是逐段生成的，在流里表现为 input_json_delta，SDK 的 stream 上下文里也能拿到，长参数场景可以边收边拼。

四、中转选型：这两块最容易露馅

同样是"支持 Claude"，不同中转在流式和工具调用上的完成度差很多。选的时候重点验这几条：

SSE 是不是真透传：前面说的掐秒表法，长回复看首字何时到。假流式一试便知。
工具调用是否完整支持：有些中转对 tools 参数支持不全，多工具、并行工具调用、或者工具流式增量就出问题。拿上面那段代码跑一遍最实在。
是否走官方协议而非逆向：走官方 /v1/messages 协议的，Anthropic 一更新字段（比如新的事件类型、新的 stop_reason）不容易挂；逆向反代某些客户端的，官方一动就可能集体翻车。
模型是否保真：别拿小模型冒充你请求的型号，工具调用对模型能力敏感，掉包了工具选择质量立刻下降。

下面这张表是我自己选型时的对照维度：

维度	官方直连	部分野中转	走官方协议的中转（如 KingFlow）
SSE 流式透传	原生	时有假流式	原样透传
Function Calling	完整	支持参差	完整支持
首字延迟(国内)	受网络/风控影响	不稳定	通常一两秒起
协议随官方更新	天然同步	逆向易挂	跟随官方协议
国内支付/对账	门槛高	看运气	后台可查用量

五、用 KingFlow 跑流式 + 工具调用

上面所有代码，我都是把 base_url 指到 https://www.kingflow.ai/v1、填上 Key 直接跑通的。对开发者来说省心的地方在于：

流式和 Function Calling、Vision 都支持，本文两段代码原样能用，不用为中转改业务逻辑；
走官方 /v1/messages 协议，不是逆向反代，流式事件类型和工具调用字段跟官方一致，SDK 不用打补丁；
一个 Key 多模型，model 从 claude-sonnet-4-6 换成 claude-opus-4-8（旗舰、适合大重构）或 claude-haiku-4-5（高频低成本）只改一个参数；
国内直连，首字延迟通常在一两秒这个量级，长连接的流式不容易被中途掐断；
后台能查用量和调用明细，倍率透明、方便对账；新人注册一般送额度，可以先拿上面的代码测通再决定充值。

鉴权就是标准两件套：ANTHROPIC_AUTH_TOKEN 放 Key，ANTHROPIC_BASE_URL 指向端点，Claude Code、Cursor 里同理配一下就能接管。

六、FAQ

Q1：中转开了 stream=True 却没有打字机效果，怎么回事？ 大概率是中转没真透传 SSE，在后端收完整段才返回。用长回复请求掐秒表验第一个 chunk 到达时间，和非流式一比就清楚。走官方协议原样透传的中转不会有这问题。

Q2：工具调用返回后模型不理我的结果？ 检查 tool_result 里的 tool_use_id 是否和模型返回的 tool_use 块 id 一致，以及包含 tool_use 的那条 assistant 消息有没有完整追加进 messages。这两点是新手最常漏的。

Q3：流式模式下能用 Function Calling 吗？ 可以。工具的参数在流里以 input_json_delta 逐段返回，用 SDK 的 stream 上下文遍历事件即可边收边拼，长参数场景尤其有用。

Q4：换成 KingFlow 后，代码要大改吗？ 基本不用。核心就是把 base_url 指到 https://www.kingflow.ai/v1、Key 换成中转的 Key，业务逻辑、SDK 调用方式、流式和工具调用的处理代码全都照旧。改模型也只是换 model 参数一个字段。