• 欢迎关注我的微信公众号“ Falost ” 右边扫描关注 --->>

Function Calling 进阶：并行工具调用、流式输出与多层 Agent 架构

AI / 大模型 神棍 81℃ 0评论

📌 回顾

上篇文章我们聊了 Function Calling 的基本用法——定义工具 schema、传给 API、解析返回的 tool_calls、执行函数并把结果喂回模型。我们写了一个能查天气、做计算、获取时间的助手，跑通了单工具调用的完整链路。

但那个版本还有个硬伤：一次只能调用一个工具。如果用户问「北京今天几度，再帮我算一下 2.5 的 3 次方」，你得先调天气，等返回了再发一次请求调计算器——两轮对话才完成一个简单的双请求。能不能一次搞定？

能。这就是今天要聊的并行工具调用（Parallel Function Calling），再加上流式输出和多层 Agent 架构，把上篇的单工具助手升级成真正的多工具协作系统。

一、并行工具调用：一次聊三件事

OpenAI 从 2024 年中（gpt-4o-2024-05-13 开始）原生支持了并行工具调用。原理很简单：模型在你的 tools 列表里判断，如果多个工具可以并行执行（互不依赖），它就把所有调用一次性返回。

来看一个完整例子。定义两个工具——一个查天气，一个做计算：

import json
from openai import OpenAI

client = OpenAI()

tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "获取指定城市的当前温度",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "城市名"}
                },
                "required": ["city"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "calculator",
            "description": "做四则运算和乘方运算",
            "parameters": {
                "type": "object",
                "properties": {
                    "expression": {
                        "type": "string",
                        "description": "数学表达式，如 2.5**3"
                    }
                },
                "required": ["expression"]
            }
        }
    }
]

messages = [
    {"role": "user", "content": "北京今天几度？再帮我算 2.5 的 3 次方"}
]

response = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools,
    tool_choice="auto"
)

msg = response.choices[0].message
print(f"模型返回了 {len(msg.tool_calls)} 个工具调用")
for tc in msg.tool_calls:
    args = json.loads(tc.function.arguments)
    print(f"  → 工具: {tc.function.name}, 参数: {args}")

运行这段代码，你会看到模型一次性返回了两个 tool_calls：

模型返回了 2 个工具调用
  → 工具: get_weather, 参数: {"city": "北京"}
  → 工具: calculator, 参数: {"expression": "2.5**3"}

拿到结果后，把两个函数的返回值同时塞回 messages，模型就会把答案综合起来——全程一次用户请求 + 一次 API 往返就搞定，不需要我们自己在外层做两轮循环。

我的做法是维护一个 tool_call 映射表，先把所有 tool_calls 按 id 存起来，并行执行（用普通的 Python 函数或者 asyncio），然后把所有结果一起追加到 messages 里：

available_functions = {
    "get_weather": get_weather,
    "calculator": calculator,
}
tool_results = []
for tc in msg.tool_calls:
    func = available_functions[tc.function.name]
    args = json.loads(tc.function.arguments)
    result = func(**args)
    tool_results.append({
        "role": "tool",
        "tool_call_id": tc.id,
        "content": str(result)
    })

messages.append(msg)
messages.extend(tool_results)

final = client.chat.completions.create(
    model="gpt-4o",
    messages=messages,
    tools=tools
)
print(final.choices[0].message.content)

到这里，并行工具调用的核心流程就跑通了。

二、流式输出 + 工具调用：分片累积的正确姿势

上篇文章我们没聊流式。但如果要做对话助手，不流式输出用户体验很差——用户发送后要等好几秒才看到完整回复。流式输出（stream=True）能让模型一边生成一边吐出 token，实时看到文字出现。

当流式输出遇到工具调用，情况就变得微妙了。模型的回复可能是文本，也可能是 tool_calls。如果 stream=True，tool_calls 是以 delta 的形式分片返回的——每个 chunk 的 delta.tool_calls 是一个列表，里面用 index 来区分属于第几个并行调用。

下面是我项目里用的流式处理模板，直接搬进代码里就能用：

def stream_and_collect_tools(client, messages, tools):
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        tools=tools,
        stream=True
    )

    collected_message = {"role": "assistant", "content": ""}
    collected_tool_calls = {}  # index => tool_call delta

    for chunk in response:
        delta = chunk.choices[0].delta if chunk.choices else None
        if not delta:
            continue

        # 收集文本
        if delta.content:
            collected_message["content"] += delta.content
            print(delta.content, end="", flush=True)

        # 收集 tool_calls（流式分段到达）
        if delta.tool_calls:
            for tc_delta in delta.tool_calls:
                idx = tc_delta.index
                if idx not in collected_tool_calls:
                    collected_tool_calls[idx] = {
                        "id": "",
                        "function": {"name": "", "arguments": ""}
                    }
                if tc_delta.id:
                    collected_tool_calls[idx]["id"] = tc_delta.id
                if tc_delta.function:
                    if tc_delta.function.name:
                        collected_tool_calls[idx]["function"]["name"] += tc_delta.function.name
                    if tc_delta.function.arguments:
                        collected_tool_calls[idx]["function"]["arguments"] += tc_delta.function.arguments

    # 组装完整的 tool_calls
    if collected_tool_calls:
        collected_message["tool_calls"] = [
            {
                "id": v["id"],
                "type": "function",
                "function": {
                    "name": v["function"]["name"],
                    "arguments": v["function"]["arguments"]
                }
            }
            for v in collected_tool_calls.values()
        ]
        print()

    return collected_message

关键点在于：tc_delta.index 是唯一能区分不同并行调用的标识，必须用它做 key。function.name 只会在第一个 chunk 里出现，后续的 chunk 只有 arguments——所以要用 += 累加，不能用赋值。

这个模板我直接用在生产里的，跑了几个月没出过问题。

三、多层 Agent 架构：从单工具到多 Agent 协作

并行工具调用解决了一次问多个独立问题，但还有一个更麻烦的场景：用户问的问题可能需要多步推理，单靠一个函数调用解决不了。

比如用户说：「帮我查一下哪家北京的公司在做向量数据库，然后把它的核心产品介绍总结一下。」这里你至少需要两个工具：先搜索公司列表，搜索结果出来后再查某家公司的产品详情。第二步依赖第一步的结果——不能并行，要串行。单工具循环可以做到（上篇的 while True 循环），但代码很快就会变成一团乱麻。

我的做法是用多层 Agent 架构。说白了就一个 Supervisor 加多个 Specialist：

  • Supervisor Agent — 接收用户请求，分析需要哪些能力，生成执行计划
  • Search Agent — 只有搜索相关工具（网页搜索、API 调用）
  • Analysis Agent — 文本理解、总结、翻译
  • Code Agent — 代码生成、执行、数据抓取

每个 Specialist Agent 本身就是一个完整的 Function Calling 循环——有自己的 tools、自己的 system prompt、独立的 messages 上下文。Supervisor 负责把任务拆解给合适的 Agent，拿到结果再决定下一步。

用代码实现时，我参考了 LangGraph 的 Multi-Agent 模式。每个 Agent 是一个节点，Supervisor 是路由节点，在它们之间传递状态：

class Agent:
    def __init__(self, name, system_prompt, tools, client):
        self.name = name
        self.system_prompt = system_prompt
        self.tools = tools
        self.client = client

    def run(self, task: str) -> str:
        messages = [
            {"role": "system", "content": self.system_prompt},
            {"role": "user", "content": task}
        ]
        while True:
            response = self.client.chat.completions.create(
                model="gpt-4o",
                messages=messages,
                tools=self.tools
            )
            msg = response.choices[0].message
            messages.append(msg)
            if not msg.tool_calls:
                return msg.content
            for tc in msg.tool_calls:
                result = execute_tool(tc)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": str(result)
                })

class Supervisor:
    def __init__(self, agents: dict, client):
        self.agents = agents
        self.client = client

    def run(self, user_request: str) -> str:
        # 分析请求 -> 选择 Agent -> 委托 -> 检查结果 -> 继续或完成
        pass  # 具体实现见 LangGraph 教程

生产级的 Supervisor 还要考虑最大迭代次数、错误重试、上下文窗口管理等。上面只是核心架子。完整的实现可以参考 LangGraph 的 Multi-Agent 教程（文末有链接）。

总结：从单工具到多 Agent 协作

今天这三个进阶点，本质是同一个问题的不同解法：怎么让大模型用好工具。

  • 并行工具调用——解决一次问多个独立问题，模型一次性返回所有调用
  • 流式输出 + 工具调用——解决用户体验问题，用 delta 累积方案处理流式中的 tool_calls
  • 多层 Agent 架构——解决复杂多步推理，用 Supervisor + Specialist 拆分任务

从上篇的单工具循环到今天的多 Agent 协作，整个体系已经能覆盖大部分生产场景了。但还有一个问题我一直在琢磨：如果你手上有十几个 Agent 工具，怎么让 Supervisor 高效地从中做选择？怎么让 Agent 之间共享上下文而不超出 token 限制？

下期预告：Agent 通信与编排框架——LangGraph、AutoGen、CrewAI 方案对比与实践

从单 Agent 到多 Agent 协作网络，不同编排框架的选型取舍，以及我用 LangGraph 搭的一个实战案例。

参考资料

转载请注明：Falost的小窝 » Function Calling 进阶：并行工具调用、流式输出与多层 Agent 架构

如果你觉得这篇文章不错或者对你有帮助，想请我喝一杯咖啡，可以打赏
打赏
喜欢 (0)
发表我的评论
取消评论

表情

Hi，您需要填写昵称和邮箱！

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址