Function Calling 用起来不难。难的是三个细节一起出现的时候。结构化输出、并行调用、流式 tool_calls,单拎出来都不新鲜。串到一条链上就全是坑。顺序错一步,后面全是补丁。代码我自己写,也真跑过。

模型只给意图,跑不跑是你的事

先把边界说清楚。Function Calling 是模型发一个结构化的调用请求,函数名加 JSON 形参。真正执行是运行时的事。模型自己不碰外部系统。这个能力 OpenAI 在 2023 年 6 月首发,挂在 gpt-3.5-turbo-0613 和 gpt-4-0613 上。现在各家都支持,Anthropic 叫 tool use、Google 叫 functionCall,字段名不一样,机制却一样。模型按你的 JSON Schema 填参数,你执行,把结果当 tool 消息喂回去,模型再接着生成。

结构化输出这里有个分叉,别搞混。tool_choice 加 additionalProperties: false 加 required 全列,能让模型几乎必定调某个函数。但这和 response_format 的 json_schema 不是一回事。后者让最终回复本身按 schema 约束解码,OpenAI 管这叫 Structured Outputs,走约束解码,更硬。想要函数参数绝对合法,用前者加全字段 required 就够。想要最终答案的 JSON 绝对合法,才上后者。

这里还有个容易漏的事实。模型在 API 之间是无状态的。它记不住上一轮。全靠你每次把完整对话历史传回去。所以 agent loop 是:发消息加工具 schema,收到 tool_calls,逐个执行,把结果当 tool 消息追加回去,再发一次。循环到模型给出纯文本为止。状态管理是你的活,不是模型的。对话一长,这段历史逼近上下文窗口,就得摘要或裁剪。

流式里 tool_calls 是分片

流式是坑最密的地方。开 stream=True 之后,模型的 tool_calls 不是一次给全。它切成 delta.tool_calls 一块一块吐出来。同一段 arguments 可能跨好几个 chunk。你还不能中途执行它。所以得先把分片收齐、拼完整,再 parse。别急。我写了一段拼分片的逻辑,真跑过:

def assemble(delta_list):
    acc = {}
    for chunk in delta_list:
        for d in chunk:
            i = d.get("index")
            slot = acc.setdefault(i, {"id": "", "name": "", "arguments": ""})
            if d.get("id"):
                slot["id"] = d["id"]
            fn = d.get("function", {})
            if fn.get("name"):
                slot["name"] = fn["name"]
            slot["arguments"] += fn.get("arguments", "")
    return [slot for _, slot in sorted(acc.items())]

注意按 index 分组,不是按 id。多个 tool call 同时流式回来时,分片是交错着的。靠 index 才能归位。我把两个城市的天气查询拆成五个分片喂进去,拼出来是两个完整的 call。arguments 是合法的 {"city":"Tokyo"} 这种。这步没坑。但漏了按 index 分组,两个调用会串成一个。

我一开始没想明白流式为什么这么别扭,以为 arguments 一整段先到、然后才轮到下一个调用,结果一看是交错的分片。这个预期跟实际不一样的地方,恰恰是流式里最容易写错的点。按一次一个调用的假设去写,就会丢参数。所以这段 assemble,我建议写进工具函数之前先当纯函数测一遍。

并行调用用 asyncio

模型一次返回多个 tool_calls,说明这几个调用彼此独立。这时候串着跑就是浪费。N 个调用就是 N 倍延迟。正确做法是并发。我跑了一段对比,两个各 2 秒的调用:

import asyncio

async def run(name, sec):
    await asyncio.sleep(sec)
    return f"{name}: {sec}s"

async def main():
    calls = [("get_weather", 2.0), ("get_weather", 2.0)]
    for c in calls:
        await run(*c)                               # 串行
    await asyncio.gather(*[run(*c) for c in calls]) # 并行

asyncio.run(main())

我这台机器上,串行 4.00 秒,并行 2.00 秒,正好压到最长那一个。数字不唬人。它印证的是那条常识:独立的调用要 gather,依赖的才串行。依赖判断错了,并行会把顺序搞乱,这是另一类 bug。

并行不是无脑 gather。三个调用里两个查天气、一个用天气结果订机票,订机票那个就得等天气回来。这时应该两个天气并行,机票串在天气后面。判断依赖看参数。B 的入参来自 A 的返回值,A 就必须先跑。模型的 tool_calls 列表本身不告诉你依赖关系。它只保证这次想调这几个。谁先谁后,你自己读参数定。

我的取舍

三条线里,流式拼接最容易漏。因为工具函数本身没问题,问题在收这一步。所以我的建议是:先把 assemble 写对、写个单测锁住,再谈并行和结构化。工具的 schema 也不是越大越好。第三方实测,一个 verbose 的 schema 平均 150 到 400 token。15 个工具就是 2000 到 5000 token 的纯开销。描述压缩到 20 个词以内能省一大截。参数能并行的并行,能流式的流式,schema 能短就短。这就是 Function Calling 全部的生产经验。

再补一个选型判断。单次问答、工具就一两个的,别上流式。第三方实测,流式只在第二轮的合成阶段才有价值,首字延迟能掉到 300 到 500ms。工具一大把、又要控 token 开销的,才值得做工具的动态注入,只把跟当前意图相关的几个工具塞进去。先跑通串行,再优化并行。别一上来就写 asyncio 把自己绕进去。