大模型流式输出与函数调用实战：从 SSE 到 Function Calling 的生产级方案

用户等待 3 秒看不到响应就会离开。本文用完整可运行代码演示如何用 SSE 实现大模型流式输出、如何用 Function Calling 让模型安全调用外部 API，以及如何把两者组合成生产级 AI 对话后端。

一、流式输出为什么是刚需

传统 API 调用是"一问一答"模式——请求发出后，必须等模型生成完毕才能拿到完整响应。当回复长达 500 字时，用户要干等 5-8 秒，体验极差。

流式输出（Streaming）改变了这个模式：模型每生成一个 Token 就立即推送给客户端，用户看到的是"逐字打字"效果，首字延迟可从 5 秒降到 300 毫秒以内。

核心协议：SSE（Server-Sent Events）

SSE 是基于 HTTP 的单向推送协议，浏览器原生支持，无需 WebSocket 的握手开销。大模型 API（OpenAI、Anthropic、通义千问）均采用 SSE 格式推送流式数据。

二、流式输出完整实现

2.1 后端：FastAPI + OpenAI Streaming

import json
import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from openai import OpenAI

app = FastAPI()
client = OpenAI(api_key="sk-your-key")

async def stream_chat(user_message: str):
    """生成器：逐 Token 产出 SSE 数据"""
    stream = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "你是有帮助的AI助手，回答简洁。"},
            {"role": "user", "content": user_message},
        ],
        stream=True,  # 关键参数：开启流式
    )

    for chunk in stream:
        # 每个 chunk 包含增量内容
        delta = chunk.choices[0].delta.content
        if delta:
            # SSE 格式：data: <json>\n\n
            yield f"data: {json.dumps({'content': delta}, ensure_ascii=False)}\n\n"

    # 发送结束标记
    yield "data: [DONE]\n\n"

@app.get("/chat")
async def chat(q: str):
    return StreamingResponse(
        stream_chat(q),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",  # Nginx 关闭缓冲，确保实时推送
        },
    )

# 启动: uvicorn main:app --host 0.0.0.0 --port 8000

2.2 前端：原生 EventSource 消费

async function streamChat(question) {
  const response = await fetch(`/chat?q=${encodeURIComponent(question)}`);

  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n");
    buffer = lines.pop(); // 保留不完整的行

    for (const line of lines) {
      if (!line.startsWith("data: ")) continue;
      const data = line.slice(6);
      if (data === "[DONE]") {
        console.log("流式结束");
        return;
      }
      const parsed = JSON.parse(data);
      // 逐字追加到页面
      document.getElementById("output").textContent += parsed.content;
    }
  }
}

// 调用
streamChat("用三句话解释什么是量子纠缠");

2.3 实操步骤

后端设置 stream=True：OpenAI SDK 会返回迭代器而非完整对象
包装为 SSE 格式：每个数据块用 data: <json>\n\n 包装
关闭代理缓冲：Nginx 需加 proxy_buffering off，否则数据会积攒后批量发送
前端流式解析：用 ReadableStream 逐块读取，注意处理跨 chunk 的不完整行
错误处理：网络中断时 EventSource 会自动重连，需用 onerror 防止无限重连

三、Function Calling：让模型安全调用外部世界

3.1 原理

Function Calling 不是模型直接执行代码，而是模型根据用户意图输出一个结构化的函数调用请求，由你的后端代码执行真正的函数，再把结果返回给模型继续推理。

用户 → 模型(决定调用查天气函数) → 后端(执行查天气) → 模型(基于结果回答) → 用户

3.2 完整代码：天气查询 Function Calling

import json
from openai import OpenAI

client = OpenAI(api_key="sk-your-key")

# 第一步：定义可用的函数
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "查询指定城市的实时天气",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，如：广州、深圳",
                    },
                    "unit": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "温度单位，默认 celsius",
                    },
                },
                "required": ["city"],
            },
        },
    }
]

# 第二步：真正的函数实现（模拟）
def get_weather(city: str, unit: str = "celsius") -> dict:
    """实际项目中这里调用天气 API"""
    mock_data = {
        "广州": {"temp": 33, "condition": "多云", "humidity": 78},
        "深圳": {"temp": 31, "condition": "晴", "humidity": 70},
    }
    result = mock_data.get(city, {"temp": 25, "condition": "未知", "humidity": 50})
    if unit == "fahrenheit":
        result["temp"] = result["temp"] * 9 / 5 + 32
    return {"city": city, **result, "unit": unit}

# 第三步：完整的对话循环（支持多轮工具调用）
def chat_with_tools(user_message: str) -> str:
    messages = [
        {"role": "system", "content": "你是天气助手，可以查询城市天气。"},
        {"role": "user", "content": user_message},
    ]

    while True:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            tools=tools,
            tool_choice="auto",  # auto: 模型自行决定是否调用
        )

        msg = response.choices[0].message
        messages.append(msg)

        # 如果模型没有调用工具，说明已经生成最终回答
        if not msg.tool_calls:
            return msg.content

        # 执行模型请求的函数调用
        for tool_call in msg.tool_calls:
            func_name = tool_call.function.name
            func_args = json.loads(tool_call.function.arguments)

            print(f"[调用函数] {func_name}({func_args})")

            # 路由到真正的函数
            if func_name == "get_weather":
                result = get_weather(**func_args)
            else:
                result = {"error": f"未知函数: {func_name}"}

            # 把函数结果返回给模型
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": json.dumps(result, ensure_ascii=False),
            })
        # 循环继续：模型拿到结果后会生成最终回答或继续调用工具

# 测试
answer = chat_with_tools("广州和深圳今天哪个更热？用华氏度告诉我")
print(f"\n回答: {answer}")

四、流式 + Function Calling 组合方案

生产环境中，你通常需要同时支持流式输出和工具调用。核心挑战是：流式模式下函数调用的参数也是分块到达的，需要先缓冲完整再执行。

async def stream_chat_with_tools(user_message: str):
    messages = [
        {"role": "system", "content": "你是智能助手。"},
        {"role": "user", "content": user_message},
    ]

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

        # 流式模式下需要手动累积
        content_buffer = ""
        tool_calls_buffer = {}  # {index: {id, name, arguments}}

        for chunk in stream:
            delta = chunk.choices[0].delta

            # 累积文本内容，实时推送
            if delta.content:
                content_buffer += delta.content
                yield f"data: {json.dumps({'content': delta.content}, ensure_ascii=False)}\n\n"

            # 累积工具调用参数（分块到达）
            if delta.tool_calls:
                for tc in delta.tool_calls:
                    idx = tc.index
                    if idx not in tool_calls_buffer:
                        tool_calls_buffer[idx] = {
                            "id": tc.id,
                            "name": tc.function.name,
                            "arguments": "",
                        }
                    if tc.function.arguments:
                        tool_calls_buffer[idx]["arguments"] += tc.function.arguments

        yield "data: [DONE]\n\n"

        # 流结束后，检查是否有工具调用
        if not tool_calls_buffer:
            break  # 没有工具调用，对话结束

        # 执行工具调用
        messages.append({"role": "assistant", "content": content_buffer})
        for idx in sorted(tool_calls_buffer):
            tc = tool_calls_buffer[idx]
            args = json.loads(tc["arguments"])
            print(f"[工具调用] {tc['name']}({args})")

            result = get_weather(**args) if tc["name"] == "get_weather" else {}

            messages.append({
                "role": "assistant",
                "tool_call_id": tc["id"],
                "content": json.dumps(result, ensure_ascii=False),
            })
        # 循环继续：模型拿到结果后流式输出最终回答

五、常见问题 FAQ

Q1：流式输出和 WebSocket 有什么区别？该用哪个？

SSE 是单向推送（服务端→客户端），基于 HTTP，浏览器原生支持，适合大模型流式场景。WebSocket 是双向通信，适合需要客户端实时推送的场景（如协作编辑）。大模型对话用 SSE 足够，且更简单。

Q2：Nginx 反向代理后流式输出变成了一次性返回怎么办？

三个关键配置：proxy_buffering off（关闭缓冲）、proxy_cache off（关闭缓存）、proxy_set_header X-Accel-Buffering no（后端主动告知关闭）。缺少任何一个都可能导致流式失效。

Q3：Function Calling 的函数参数解析失败了怎么办？

流式模式下参数分块到达，必须等流结束后再 json.loads。非流式模式下偶发解析失败，建议加 try/except 并把原始 arguments 记入日志，用重试机制兜底。

Q4：如何限制模型只调用我允许的函数？

用 tool_choice 参数：设为 {"type": "function", "function": {"name": "get_weather"}} 可强制只调指定函数；设为 "none" 可完全禁用工具调用。生产环境建议用 allowlist 做二次校验。

Q5：流式输出怎么做 Token 计数和成本统计？

流式模式下 API 不返回 usage 字段。两种方案：① 用 tokenizer 库（如 tiktoken）在服务端实时计数；② 流结束后发一个 stream_options={"include_usage": True} 参数请求，OpenAI 会在最后一个 chunk 返回用量统计。

六、总结

流式输出解决的是用户体验问题（首字延迟），Function Calling 解决的是能力边界问题（调用外部数据）。两者组合是 2026 年生产级 AI 应用的标配架构：

能力	技术	关键指标
实时响应	SSE 流式输出	首字延迟 < 500ms
外部数据	Function Calling	工具调用准确率 > 95%
组合方案	流式 + 工具调用缓冲	需处理分块参数累积

落地建议：先用非流式 + Function Calling 跑通业务逻辑，验证函数调用准确率达标后再叠加流式输出优化体验。别一上来就上组合方案——调试分块参数累积的痛苦会让你怀疑人生。

一、流式输出为什么是刚需#

核心协议：SSE（Server-Sent Events）#

二、流式输出完整实现#

2.1 后端：FastAPI + OpenAI Streaming#

2.2 前端：原生 EventSource 消费#

2.3 实操步骤#

三、Function Calling：让模型安全调用外部世界#

3.1 原理#

3.2 完整代码：天气查询 Function Calling#

四、流式 + Function Calling 组合方案#

五、常见问题 FAQ#

六、总结#