实际场景中，大模型生成内容可能需要几秒时间，「一次性返回」会让用户感觉卡顿。这一篇我们就来学习流式输出：让LLM生成内容的过程中，实时把内容推送给用户，实现「边生成、边展示」的流畅体验。
一、什么是流式输出？
1. 传统非流式输出（我们之前的方式）
- • 过程：用户发指令 → 程序等待LLM生成完整结果 → 一次性返回所有内容
- • 体验：等待时间长，页面/终端无响应，像「卡了一样」
- • 类比：点外卖后，等所有菜做好才一次性送到家
2. 流式输出（我们要实现的方式）
- • 过程：用户发指令 → LLM生成一点内容就返回一点 → 程序实时接收并展示
- • 体验：实时看到内容生成，无卡顿感，交互更友好
- • 类比：点外卖后，菜做好一道送一道，不用等全部做完
核心差异
| 维度 | 非流式输出 | 流式输出 |
|---|---|---|
| 数据返回方式 | 一次性返回完整结果 | 分批次返回片段结果 |
| 等待体验 | 全程等待，无中间反馈 | 实时反馈，边等边看 |
| 资源占用 | 需缓存完整结果后再处理 | 实时处理片段，内存占用低 |
| 适用场景 | 短文本、快速计算 | 长文本、复杂推理、交互型Agent |
二、为什么Agent需要流式输出？
对我们的四则运算Agent来说，虽然计算结果短，但流式输出是工业级Agent的必备能力：
- 1. 提升用户体验：哪怕只等0.5秒，实时输出也比空白等待更友好；
- 2. 适配复杂场景：后续扩展Agent到多轮对话、长文本生成时，流式输出是基础；
- 3. 实时监控状态：能看到LLM的思考/生成过程，便于调试问题；
- 4. 降低超时风险：分段返回比一次性返回更不容易触发超时。
三、实战：给工具调用型Agent加流式输出
我们基于第三篇的工业级Agent代码，只改「LLM调用部分」，其余逻辑完全复用，保证极简易懂。
3.1 核心改造思路
- 1. 把
CreateChatCompletion（非流式）换成CreateChatCompletionStream（流式）； - 2. 遍历流式响应，实时接收每一段内容；
- 3. 解析流式返回的工具调用指令，保持原有工具调用逻辑；
- 4. 实时打印/展示生成过程。
3.2 完整流式输出版Agent代码
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"os"
"strings"
"time"
"github.com/sashabaranov/go-openai"
)
// ==================== 1. Agent 核心结构体（新增耗时统计字段） ====================
// Agent 保存 Agent 运行时的状态与记忆，并记录流式调用的耗时指标。
type Agent struct {
UserPrompt string // 用户自然语言指令
Memory string // 工具执行结果记忆
FinalAnswer string // 最终输出答案
FirstTokenLatency int64 // 首 Token 耗时（毫秒）
TotalLatency int64 // 总耗时（毫秒）
}
// ==================== 2. 四则运算工具 ====================
// parseCalculation 从文本解析「数字1 运算符 数字2」。
func parseCalculation(text string) (num1 float64, op rune, num2 float64, err error) {
_, err = fmt.Sscanf(strings.TrimSpace(text), "%f %c %f", &num1, &op, &num2)
return num1, op, num2, err
}
// calculator 执行加减乘除四则运算，返回可读的计算结果字符串。
func calculator(expression string) string {
num1, op, num2, err := parseCalculation(expression)
if err != nil {
return fmt.Sprintf("解析失败：%v", err)
}
var res float64
switch op {
case '+':
res = num1 + num2
case '-':
res = num1 - num2
case '*':
res = num1 * num2
case '/':
if num2 == 0 {
return "错误：除数不能为0"
}
res = num1 / num2
default:
return fmt.Sprintf("不支持运算符：%c", op)
}
return fmt.Sprintf("%.2f %c %.2f = %.2f", num1, op, num2, res)
}
// ==================== 3. 工具定义：给云端 LLM 看的工具清单 ====================
// agentTools 声明 Agent 可用的工具，云端 LLM 据此自主决策调用。
var agentTools = []openai.Tool{
{
Type: openai.ToolTypeFunction,
Function: &openai.FunctionDefinition{
Name: "calculator",
Description: "四则运算计算器，支持加减乘除计算，其他运算不支持",
Parameters: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"expression": map[string]interface{}{
"type": "string",
"description": "数学计算表达式，格式：数字 + 空格 + 运算符 + 空格 + 数字",
},
},
"required": []string{"expression"},
},
},
},
}
// ==================== 4. 核心改造：带耗时统计的流式 LLM 调用 ====================
// callCloudLLMStream 以流式方式调用云端大模型，并统计首 Token 与总耗时。
// 返回合并后的完整消息（含 Content 与 ToolCalls）。
func (a *Agent) callCloudLLMStream() (openai.ChatCompletionMessage, error) {
// 初始化客户端（复用原有配置）
apikey := os.Getenv("DEEPSEEK_API_KEY")
if apikey == "" {
return openai.ChatCompletionMessage{}, fmt.Errorf("DEEPSEEK_API_KEY 环境变量未设置")
}
cfg := openai.DefaultConfig(apikey)
cfg.BaseURL = "https://api.deepseek.com"
client := openai.NewClientWithConfig(cfg)
// 构建消息（复用标准化 Prompt）
messages := []openai.ChatCompletionMessage{
{
Role: openai.ChatMessageRoleSystem,
Content: `### 角色定位
你是一个「工具调用型四则运算Agent」，核心能力是解析自然语言→调用calculator工具→返回结果。
### 核心任务
必须通过tool_calls调用calculator工具，禁止直接返回计算结果。
### 工具规则
expression参数格式：数字 + 空格 + 运算符 + 空格 + 数字（如150 + 250）。
### 输出格式
符合条件则返回tool_calls，否则返回固定提示语。`,
},
{
Role: openai.ChatMessageRoleUser,
Content: a.UserPrompt,
},
}
// 1. 记录请求开始时间（总耗时起点）
startTotalTime := time.Now()
// 2. 创建流式请求
req := openai.ChatCompletionRequest{
Model: "deepseek-v4-flash", // DeepSeek 官方模型名，第三方模型替换对应名称
Messages: messages,
Tools: agentTools,
Temperature: 0.1,
Stream: true, // 开启流式输出
}
stream, err := client.CreateChatCompletionStream(context.Background(), req)
if err != nil {
return openai.ChatCompletionMessage{}, fmt.Errorf("流式调用失败：%w", err)
}
defer stream.Close() // 关闭流，避免资源泄漏
fmt.Println("\n【1. 推理】云端大模型实时生成中...")
var finalMsg openai.ChatCompletionMessage
firstTokenReceived := false // 标记是否收到第一个有效 Token
for {
// 逐段接收 LLM 返回的内容
response, err := stream.Recv()
if errors.Is(err, io.EOF) {
break // 流结束
}
if err != nil {
// 非 EOF 错误直接返回，避免继续读取已损坏的流
return finalMsg, fmt.Errorf("流式接收失败：%w", err)
}
if len(response.Choices) == 0 {
continue
}
// 3. 统计首 Token 耗时（仅第一次收到有效数据时记录）
if !firstTokenReceived {
a.FirstTokenLatency = time.Since(startTotalTime).Milliseconds()
firstTokenReceived = true
fmt.Printf("【耗时统计】首 Token 耗时：%d 毫秒\n", a.FirstTokenLatency)
}
// 合并片段内容
delta := response.Choices[0].Delta
// 合并工具调用指令
if len(delta.ToolCalls) > 0 {
if len(finalMsg.ToolCalls) == 0 {
finalMsg.ToolCalls = delta.ToolCalls
} else {
// 累加后续片段的 arguments
finalMsg.ToolCalls[0].Function.Arguments += delta.ToolCalls[0].Function.Arguments
}
}
// 合并文本内容（边界约束提示语）
if delta.Content != "" {
fmt.Println(delta.Content) // 实时打印生成的内容
finalMsg.Content += delta.Content
}
}
// 4. 统计总耗时（毫秒）
a.TotalLatency = time.Since(startTotalTime).Milliseconds()
fmt.Printf("\n【耗时统计】总耗时：%d 毫秒\n", a.TotalLatency)
fmt.Println("【1. 推理】大模型生成完成")
return finalMsg, nil
}
// ==================== 5. ReAct 循环（新增耗时展示） ====================
// Run 运行 ReAct 循环：流式推理 → 工具调用/边界约束 → 记忆闭环（含耗时统计）。
func (a *Agent) Run() {
fmt.Println("===== 带耗时统计的流式 Agent 启动 =====")
fmt.Printf("用户指令：%s\n", a.UserPrompt)
// 1. 推理：调用流式 LLM 接口
llmMsg, err := a.callCloudLLMStream()
if err != nil {
a.FinalAnswer = fmt.Sprintf("调用失败：%v", err)
return
}
// 2. 行动：根据 LLM 结果执行工具或处理边界约束
if len(llmMsg.ToolCalls) > 0 {
// 工具调用场景
for _, toolCall := range llmMsg.ToolCalls {
fmt.Printf("【2. 行动】调用【%s】工具\n", toolCall.Function.Name)
if toolCall.Function.Name != "calculator" {
fmt.Printf("未知工具：%s\n", toolCall.Function.Name)
continue
}
// 解析参数：{"expression": "150 + 250"}
var cal struct {
Expression string `json:"expression"`
}
if err = json.Unmarshal([]byte(toolCall.Function.Arguments), &cal); err != nil {
a.Memory = fmt.Sprintf("解析失败：%v", err)
continue
}
fmt.Printf("计算表达式：%s\n", cal.Expression)
a.Memory = calculator(cal.Expression)
fmt.Printf("工具执行结果：%s\n", a.Memory)
}
} else if llmMsg.Content != "" {
// 边界约束场景：LLM 直接返回提示语
a.Memory = llmMsg.Content
fmt.Printf("【2. 边界约束】%s\n", a.Memory)
}
// 3. 完成：生成最终答案（包含耗时统计）
fmt.Println("\n【3. 完成】Agent 生成最终答案")
a.FinalAnswer = fmt.Sprintf("✅ %s | 首 Token 耗时：%dms | 总耗时：%dms",
a.Memory, a.FirstTokenLatency, a.TotalLatency)
}
// ==================== 主函数：测试带耗时统计的流式 Agent ====================
func main() {
// 测试用例1：正常四则运算
agent := &Agent{UserPrompt: "帮我算150加250等于多少"}
agent.Run()
fmt.Println("\n===== Agent 最终输出 =====")
fmt.Println(agent.FinalAnswer)
// 测试用例2：非四则运算（边界约束）
agent2 := &Agent{UserPrompt: "帮我算100的平方根"}
fmt.Println("\n------------------------")
agent2.Run()
fmt.Println("\n===== Agent 最终输出 =====")
fmt.Println(agent2.FinalAnswer)
}
四、关键代码解析
1. 开启流式开关
Stream: true, // 只需加这一行，就能开启流式输出这是流式输出的核心开关，告诉LLM「不要一次性返回结果，分批次发」。
2. 流式接收数据
stream, err := client.CreateChatCompletionStream(...)
for {
response, err := stream.Recv() // 逐段接收
if err != nil { break } // 流结束就退出
// 处理每一段数据
}- •
stream.Recv()：每次调用取一段LLM生成的内容，像「接水管里的水，接一点用一点」； - • 循环直到
Recv()返回错误（表示流结束）。
3. 合并分段内容
LLM可能把工具调用的参数（如{"expression":"150 + 250"}）分成多段返回，需要拼接起来才能正常解析：
finalMsg.ToolCalls[0].Function.Arguments += delta.ToolCalls[0].Function.Arguments4. 实时展示体验
fmt.Println(delta.Content) // 实时打印生成的内容这一行让用户能看到LLM「边想边说」的过程，是流式输出的体验核心。
五、运行效果（直观感受流式输出）
正常工具调用场景
===== 带耗时统计的流式 Agent 启动 =====
用户指令：帮我算150加250等于多少
【1. 推理】云端大模型实时生成中...
【耗时统计】首 Token 耗时：404 毫秒
【耗时统计】总耗时：1031 毫秒
【1. 推理】大模型生成完成
【2. 行动】调用【calculator】工具
计算表达式：150 + 250
工具执行结果：150.00 + 250.00 = 400.00
【3. 完成】Agent 生成最终答案
===== Agent 最终输出 =====
✅ 150.00 + 250.00 = 400.00 | 首 Token 耗时：404ms | 总耗时：1031ms
边界约束场景（实时生成文本）
------------------------
===== 带耗时统计的流式 Agent 启动 =====
用户指令：帮我算100的平方根
【1. 推理】云端大模型实时生成中...
【耗时统计】首 Token 耗时：206 毫秒
根据
我的
工具
能力
，
我
仅
支持
四
则
运算
（
加减
乘
除
），
不支持
平方
根
等
运算
。
请
提供
四
则
运算
表达式
，
例如
"
150
+
250
"。
【耗时统计】总耗时：1099 毫秒
【1. 推理】大模型生成完成
【2. 边界约束】根据我的工具能力，我仅支持四则运算（加减乘除），不支持平方根等运算。请提供四则运算表达式，例如"150 + 250"。
【3. 完成】Agent 生成最终答案
===== Agent 最终输出 =====
✅ 根据我的工具能力，我仅支持四则运算（加减乘除），不支持平方根等运算。请提供四则运算表达式，例如"150 + 250"。 | 首 Token 耗时：206ms | 总耗时：1099ms
六、流式输出核心总结
- 1. 核心开关：请求时加
Stream: true，就能开启流式； - 2. 核心方法：用
CreateChatCompletionStream替代CreateChatCompletion，循环Recv()接收分段数据； - 3. 关键操作：分段内容需要合并，尤其是工具调用的参数；
- 4. 体验核心：实时打印/展示每一段生成的内容，提升交互友好度；
- 5. 工程化要点：记得用
defer stream.Close()关闭流，避免资源泄漏。
流式输出是工业级Agent的基础能力，学会这篇内容，你的Agent就能从「卡顿的一次性响应」变成「流畅的实时响应」，适配更多实际应用场景。
