手写 Agent ⑥，流式输出、错误处理与重试

开场：十五秒的沉默，和一次 503

用 mini-agent 用了一阵子，我发现自己有个下意识的动作：问完一个问题，盯着光标看几秒，然后伸手去摸手机。

因为它太安静了。模型要把整段回答全部生成完，程序才会一下子打印出来。一个稍长的回答要十几秒，这十几秒里屏幕上什么都没有，我不知道它是在思考，还是卡住了，还是网络断了。

另一个问题更让人恼火。有一次我让它做一个要转三十多圈的重构，转到第二十几圈时，服务端返回了一个 503：服务繁忙。mini-agent 直接报错退出了这一轮，前面二十几圈的进度虽然还在历史里，但我得重新告诉它"继续"，而它常常要重新读几个文件才能想起来自己在干什么。

一个 503，只要等两秒再试一次就好了。 可是我的程序不会等。

这一篇解决这两个问题：流式输出，让文字边生成边显示；重试，让程序在遇到"等一等就好"的错误时自己等一等。还有一个更刁钻的情况：模型的输出在工具调用写到一半的时候被截断了。读完本篇，你应该能回答这几个问题：

• 流式输出在 HTTP 层面长什么样？怎么把一段段的增量拼回完整的响应？
• 工具调用的参数也是一段一段到的，怎么拼？
• 哪些错误值得重试，哪些不值得？
• 指数退避和随机抖动是什么，为什么要它们？
• 输出在工具调用中途被截断了，怎么办？
• 为什么重试模型调用是安全的，重试工具调用却不是？

一、流式输出

1.1 为什么流式更好

第 07 篇讲过，模型是一个 token 一个 token 地生成的。既然如此，为什么要等它全部生成完才显示？

流式输出就是：模型每生成一小段，服务端就推送给客户端一小段，客户端收到就立刻显示。总耗时一样，但用户的感受完全不同：第一个字在一两秒内就出现了，后面的字一个个往外蹦，你能看到它在"想"，也能在它跑偏的第一时间按下 Ctrl+C。

衡量这种感受，有两个不同的指标：首 token 延迟（从发出请求到第一个字出现）和总延迟（到最后一个字结束）。非流式时，用户只能感受到总延迟；流式时，用户首先感受到的是首 token 延迟，而它通常只占总延迟的一小部分。第 07 篇讲过，首 token 之前是 prefill，读完全部输入；之后是 decode，一个一个往外吐。流式输出没有让模型变快，它只是把等待从"一整段沉默"变成了"看着它写"。

对 Agent 来说还有一个额外的好处：你能更早地发现它跑偏了。 它刚写出"我先把整个 utils 目录重构一下"，你就可以按下 Ctrl+C，而不用等它把一大段计划写完、甚至已经开始调用工具。

1.2 Server-Sent Events

在 HTTP 层面，流式输出用的是一种很简单的格式：Server-Sent Events（SSE）。服务端返回的不是一个完整的 JSON，而是一行一行的事件，每一行以 data: 开头，后面是一小段 JSON。

DeepSeek 的流式响应（OpenAI 兼容格式）大致是这样的：

data: {"choices":[{"delta":{"role":"assistant","reasoning_content":"想一想"}}]}

data: {"choices":[{"delta":{"content":"我先"}}]}

data: {"choices":[{"delta":{"content":"读文件。"}}]}

: keep-alive

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"call_1","type":"function","function":{"name":"read_file","arguments":""}}]}}]}

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\"path\":"}}]}}]}

data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":"\"a.go\"}"}}]}}]}

data: {"choices":[{"delta":{},"finish_reason":"tool_calls"}],"usage":{"prompt_tokens":50,"completion_tokens":12}}

data: [DONE]

（这段内容来自配套仓库的测试 deepseek_test.go，就是用来模拟服务端的。）

逐行看：

• 前三行：思考内容和文字内容的增量，reasoning_content 和 content 字段里各有一小段；
• : keep-alive：以冒号开头的是注释，服务端用它来保持连接，直接忽略；
• 接下来三行：一个工具调用。第一块带着 id 和工具名，参数是空的；后面几块只有参数的片段，要按 index 拼起来：{"path": + "a.go"} = {"path":"a.go"}；
• 倒数第二行：finish_reason 告诉你为什么停下，usage 只在这里出现；
• 最后一行 [DONE]：流结束了。

1.3 手写解析

解析的代码在 internal/llm/deepseek_stream.go 里，不到一百行。核心是一个循环，一行一行地读：

for sc.Scan() {
    line := strings.TrimSpace(sc.Text())
    if !strings.HasPrefix(line, "data:") {
        continue // 空行、注释（": keep-alive"）之类
    }
    data := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
    if data == "[DONE]" {
        finished = true
        break
    }
    var ch dsChunk
    if err := json.Unmarshal([]byte(data), &ch); err != nil {
        return nil, fmt.Errorf("SSE 数据不是合法 JSON: %w", err)
    }
    // ……处理这一块
}

每一块里的增量，分三种处理：

if delta.ReasoningContent != "" {
    msg.ReasoningContent += delta.ReasoningContent
    if h.OnThinking != nil {
        h.OnThinking(delta.ReasoningContent)
    }
}
if delta.Content != nil && *delta.Content != "" {
    text.WriteString(*delta.Content)
    if h.OnText != nil {
        h.OnText(*delta.Content)
    }
}
for _, tc := range delta.ToolCalls {
    for len(calls) <= tc.Index {
        calls = append(calls, dsToolCall{Type: "function"})
    }
    dst := &calls[tc.Index]
    if tc.ID != "" {
        dst.ID = tc.ID
    }
    if tc.Function.Name != "" {
        dst.Function.Name = tc.Function.Name
        if h.OnToolUse != nil {
            h.OnToolUse(tc.Function.Name)
        }
    }
    dst.Function.Arguments += tc.Function.Arguments
}

• 文字和思考：一边累积起来，一边立刻交给 handler 显示；
• 工具调用：按 index 找到对应的那一个（没有就新建），名字和 id 记下来，参数片段拼接上去。一次回复里有多个工具调用时（第 19 篇的并行调用），它们的 index 不同，各拼各的。

流结束之后，把累积的东西组装成一个完整的 Response：

msg.Content = strPtr(text.String())
msg.ToolCalls = calls
out.Role = RoleAssistant
out.Content = fromDeepSeek(msg)
out.StopReason = finishToStop(finish)
return &out, nil

fromDeepSeek 就是第 19 篇讲的那个翻译函数，和非流式调用用的是同一个。所以流式调用最终返回的 Response，和非流式调用一模一样。 Agent 循环的其余部分完全不用关心它是不是流式的。

1.4 流断了怎么办

流式响应有一个非流式没有的问题：它可能断在中间。 网络抖了一下，或者服务端出了问题，流在 [DONE] 之前就结束了。这时候拼出来的东西是不完整的，不能当成正常的响应。

if err := sc.Err(); err != nil {
    return nil, fmt.Errorf("读取流时断开: %w", err)
}
if !finished && finish == "" {
    return nil, fmt.Errorf("流在结束之前就断了: %w", io.ErrUnexpectedEOF)
}

既没有收到 [DONE]，也没有收到 finish_reason，就返回一个错误，并且用 %w 包上 io.ErrUnexpectedEOF。下面会看到，这个错误会被判断为"值得重试"。

1.5 接进 Agent

第 24 篇定义了一个 Streamer 接口：

// Streamer 是支持流式输出的模型。Client、Demo、DeepSeek 和 RetryModel 都实现了它。
type Streamer interface {
    Stream(ctx context.Context, req *Request, h StreamHandler) (*Response, error)
}

Agent 调用模型时，如果开了流式、而且模型支持，就用流式调用，把文字增量交给 UI 的 TextDelta：

// call 调用一次模型。支持流式时边生成边把文字交给 UI，用户不用干等。
func (a *Agent) call(ctx context.Context, req *llm.Request) (*llm.Response, error) {
    // ……
    s, ok := a.Model.(llm.Streamer)
    if !a.Streaming || !ok {
        resp, err := a.Model.Create(ctx, req)
        if err == nil {
            a.UI.Text(resp.Text())
        }
        return resp, err
    }
    return s.Stream(ctx, req, llm.StreamHandler{OnText: a.UI.TextDelta})
}

终端 UI 的 TextDelta 只做一件事：fmt.Print(s)，不换行，收到一段就打印一段。mini-agent 在 main.go 里设置了 a.Streaming = true。

注意 Agent 只传了 OnText 这一个回调。StreamHandler 里其实还有 OnThinking 和 OnToolUse，解析器也会调用它们，但 mini-agent 没有用：

• 思考内容不显示：DeepSeek 的思考往往很长，全部打在终端里会淹没真正的回答。它会被完整地存进历史（第 12 篇说过，带工具时必须原样回传），只是不给人看；
• 工具名提前到了也不显示：流式响应里，工具名在第一块就到了，参数要一段段地拼。如果模型正在用 edit_file 写一个几百行的新文件，参数要传好几秒，这几秒里终端是安静的。用 OnToolUse 提前打印一行"正在准备调用 edit_file……"，体验会好很多。

这两处都是很好的小练习：在 call 里把这两个回调接上，思考内容用灰色打印、只打前几十个字，工具名一到就提示。改的都是 UI，Agent 的逻辑一行都不用动。

不设 Key 的演示模型也"支持"流式：它没有真正的生成过程，就先把整段回答准备好，再一小段一小段地交给 handler，每段之间停顿一下（DEMO_DELAY_MS 控制停顿多久）。效果上一样。

二、错误处理与重试

2.1 哪些错误值得重试

调用模型 API，会遇到各种错误。关键是分清楚：哪些是"等一等可能就好了"，哪些是"重试一万次也没用"。

internal/llm/retry.go 里的判断：

// Retryable 判断一个错误值不值得重试。
func Retryable(err error) bool {
    var apiErr *APIError
    if errors.As(err, &apiErr) {
        switch apiErr.StatusCode {
        case 408, 409, 429, 500, 502, 503, 504:
            return true
        }
        return apiErr.Type == "overloaded"
    }
    if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
        return false
    }
    var netErr net.Error
    return errors.As(err, &netErr) || errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF)
}

这里用了 Go 的 errors.As 和 errors.Is，它们能穿透用 %w 包装过的错误。所以 1.4 节那个"流在结束之前就断了: %w"的错误，虽然外面包了一层文字，errors.Is(err, io.ErrUnexpectedEOF) 依然能识别出来。

400 不重试这一条尤其重要。第 28 篇开场那个上下文超长的错误，就是 400。如果对它重试，只会重复失败五次，白白多等几十秒。

2.2 RetryModel：一个包装器

重试的逻辑没有写在 Agent 里，也没有写在 DeepSeek 客户端里，而是做成了一个包装器：

// RetryModel 给任何 Model 加上指数退避重试（第 29 篇）。
//
// 只重试"等一等可能就好了"的错误：限流（429）、服务端错误和过载（5xx）、网络中断。
// 400（请求本身有问题）、401（Key 不对）这类错误重试一万次也没用，直接返回。
type RetryModel struct {
    Model      Model
    MaxRetries int           // 默认 4
    BaseDelay  time.Duration // 默认 1 秒，之后每次翻倍
    MaxDelay   time.Duration // 默认 30 秒

    // OnRetry 在每次重试前调用，用来告诉用户"我还活着，只是在等"。
    OnRetry func(attempt int, wait time.Duration, err error)
    // ……
}

RetryModel 自己也实现了 Model 接口（和 Streamer 接口），里面包着另一个 Model。对外看，它就是一个"不太会失败"的模型。 任何模型都可以包上它，Agent 完全不知道重试的存在。

mini-agent 在 main.go 里是这么组装的：

// 第 29 篇：限流、过载、网络抖动时自动重试。
wrap := func(c *llm.Client) llm.Model {
    return &trace.Model{Tracer: tracer, Inner: &llm.RetryModel{Model: c, OnRetry: func(n int, wait time.Duration, err error) {
        fmt.Fprintf(os.Stderr, "\n\033[33m（第 %d 次重试，%.0f 秒后：%v）\033[0m\n", n, wait.Seconds(), err)
    }}}
}

一层套一层：最里面是真正的模型，外面是重试，再外面是记账（第 37 篇）。每一层只管一件事。

2.3 指数退避加随机抖动

重试的核心循环：

for attempt := 0; ; attempt++ {
    resp, err := call()
    if err == nil || attempt >= maxRetries || !Retryable(err) || ctx.Err() != nil {
        return resp, err
    }
    // 指数退避：1s, 2s, 4s, 8s……再加一点随机抖动，免得一群客户端同时醒来又一起把服务打挂。
    wait := base << attempt
    wait = wait/2 + time.Duration(rand.Int63n(int64(wait/2)+1))
    var apiErr *APIError
    if errors.As(err, &apiErr) && apiErr.RetryAfter > 0 {
        wait = apiErr.RetryAfter // 服务端说了等多久，就听它的
    }
    wait = min(wait, maxDelay)
    if r.OnRetry != nil {
        r.OnRetry(attempt+1, wait, err)
    }
    if err := sleep(ctx, wait); err != nil {
        return nil, err
    }
}

三个设计：

指数退避。 第一次等 1 秒，第二次等 2 秒，第三次 4 秒，第四次 8 秒。为什么不每次都等 1 秒？因为服务端过载时，它需要时间恢复；你等得越久，它越可能恢复了。而且如果所有客户端都每秒重试一次，服务端只会更加不堪重负。

随机抖动。 wait/2 + 随机(0, wait/2)：在指数退避的基础上，实际等待时间是"一半固定、一半随机"。想象服务端刚刚恢复，一千个客户端都在等待重试，如果它们都精确地等了 4 秒，就会在同一瞬间一起冲上来，把服务端又打挂。加一点随机，让它们错开。

听服务端的。 如果服务端在响应头里带了 retry-after（"请在 N 秒后重试"），就照它说的等。服务端比客户端更清楚自己什么时候能恢复。

还有两个上限：最多重试 4 次（默认），每次最多等 30 秒。等待本身也是可以被打断的：sleep 会同时监听 context，用户按了 Ctrl+C，等待立刻结束。

2.4 为什么重试模型调用是安全的

重试有一个前提，很容易被忘记：被重试的操作，执行两次和执行一次的效果要一样。

模型调用满足这个前提：它没有副作用。同一个请求发两次，最坏的结果是多花了一次钱，世界上什么都没有被改变。所以 RetryModel 可以放心地重试。

但工具调用不一样。edit_file 改了一个文件，run_bash 跑了 git commit，第 20 篇那个假想的 cancel_order 取消了一笔订单。这些操作如果在"执行了，但结果没传回来"的时候被重试，就会做两次。这也是为什么 mini-agent 的重试只包在模型外面，工具执行失败从不自动重试，而是交给模型去判断（下面 2.6 节）：模型看得到上一次的结果，知道那条命令到底有没有执行。

还有一件事：重试一个大请求，意味着把十几万 token 的上下文再发一遍。好在第一次请求如果已经到了服务端，前缀多半已经被缓存了，重试时能吃到缓存。但这不保证，所以第 37 篇的账本里，重试带来的费用会和正常调用记在一起，你能看到它。

2.5 流式调用的重试

// Stream 如果底层模型支持流式，就流式调用；重试时整个请求重来一遍。
func (r *RetryModel) Stream(ctx context.Context, req *Request, h StreamHandler) (*Response, error) {
    s, ok := r.Model.(Streamer)
    if !ok {
        return r.Create(ctx, req)
    }
    return r.do(ctx, func() (*Response, error) { return s.Stream(ctx, req, h) })
}

流式调用断在中间时，没法"接着上次的位置继续"，只能整个请求重来一遍。这意味着，用户可能会在屏幕上看到前半段文字出现了两次：一次是断掉的那次，一次是重试的这次。mini-agent 的做法是在重试前打印一行黄色的"第 N 次重试"，让用户知道发生了什么。

2.6 工具报错：交给模型

最后再强调一种"错误"，它和前面的完全不同：工具执行出错。 文件不存在、命令执行失败、测试没通过。这些不是程序的错误，不需要程序重试，而是要交给模型，让它自己决定怎么办。

这件事从第 25 篇起就是这么做的：注册表的 Run 永远返回一个 tool_result，出错时标上 is_error。模型看到错误信息，换个路径、换个参数、修一下代码再试。程序重试的是"网络和服务端"的错误，模型处理的是"任务本身"的错误。 两者各管各的。

2.7 超时：整个请求的上限

第 24 篇说过，HTTP 客户端设了 10 分钟的超时。这个超时管的是整个请求，从发出去，到流式响应的最后一块收完。

对非流式请求，这很合理：10 分钟还没有回音，多半是出了问题。对流式请求，它就有点粗糙了：一个正常的长回答，可能持续好几分钟都在稳定地往外吐字；而一个出了问题的连接，可能第一个字都还没到就卡住了。两者在"总时长"上没法区分。

更精细的做法是空闲超时：只要还在持续收到数据，就一直等；连续 N 秒什么都没收到，才判定为卡住。配合重试，卡住的连接会被及时断开、重新发起。mini-agent 没有做这一层，用一个宽松的总超时兜底。如果你在网络不太好的环境里用，这是一个值得加的改进。

2.8 400 也有例外：上下文超长

2.1 节说 400 不重试，因为请求本身有问题，重发一模一样的请求只会一模一样地失败。

但有一种 400 是可以"换个请求"再试的：上下文超长。 第 28 篇开场那个场景，如果 harness 能识别出"这是上下文超长的 400"，它可以先尝试压缩历史、或者把最近那个巨大的工具结果截短，然后用一个新的请求再试一次。这不是 RetryModel 该做的事，因为它只会原样重发；而是 Agent 循环该做的事，因为只有它能修改历史。

mini-agent 没有实现这一点，它靠第 28 篇的截断，在入口处就避免了这种情况发生。但如果你接的模型上下文窗口比较小，这是一个很实用的兜底。

三、刁钻的情况：截断在工具调用中间

3.1 发生了什么

第 07 篇说过，生成可能因为达到 max_tokens 上限而被截断，这时 stop_reason 是 max_tokens。如果截断发生在普通文字里，问题不大，模型的话没说完而已。

但如果截断发生在工具调用写到一半的时候呢？比如模型正在用 edit_file 写一个很大的新文件，new_str 写到一半，达到了上限。这时候：

• 工具调用的参数是不完整的。在 OpenAI 兼容格式里，arguments 是一个字符串，截断之后它不是合法的 JSON；即使碰巧是合法的，内容也是残缺的；
• 但这个 tool_use 已经在模型的回复里了，第 02 篇的铁律要求：每个 tool_use 都必须有对应的 tool_result。

3.2 怎么处理

两件事：不执行，但给结果。 agent.go 里的处理：

case "max_tokens":
    // 输出被截断。如果截断发生在工具调用里，这些调用不能执行，但必须给它们一个结果。
    if len(calls) == 0 {
        return resp.Text(), nil
    }
    truncated = true

if truncated {
    for _, call := range calls {
        results = append(results, llm.ToolResult(call.ID, "你的输出达到 max_tokens 上限被截断了，这个工具调用没有执行。"+
            "如果是要写入大段内容，请拆成多次较小的修改。", true))
    }
}

• 不执行：参数是残缺的，执行它可能把文件写成半截，比不执行更糟；
• 给每个调用一个错误结果：满足铁律，让循环能继续；
• 告诉模型怎么办："请拆成多次较小的修改"。模型下一圈通常会先新建一个小文件，再用几次 edit_file 往里加内容。

另外，在把 DeepSeek 的响应翻译回内容块时（第 19 篇的 fromDeepSeek），如果 arguments 不是合法的 JSON，会把参数换成一个空对象 {}。这样即使截断之外的原因导致参数残缺，工具也会因为"缺少必填参数"而报错，模型看到错误会重试，而不是用一堆乱码去执行。

四、动手

实验 1：看流式输出

流式输出用演示模型也能看到效果；要看真实的思考和工具调用增量，需要 export DEEPSEEK_API_KEY=sk-...。在线 Playground 跑不了 mini-agent。

go run ./cmd/mini-agent

随便问一句，看文字是不是一段一段地蹦出来。不设 Key 时，演示模型每段之间停顿 15 毫秒，可以调慢一点看得更清楚：

DEMO_DELAY_MS=100 go run ./cmd/mini-agent

接上 DeepSeek 之后，问一个需要它说一大段话的问题，在它说到一半的时候按 Ctrl+C，看看是不是立刻停下了。

实验 2：看重试

演示模型可以按概率"假装"服务端出错，专门用来观察重试：

DEMO_FAIL_RATE=0.5 go run ./cmd/mini-agent

有一半的请求会返回 429（带着 retry-after: 1）或者 503。你会在终端里看到黄色的"（第 1 次重试，1 秒后：model api 429 rate_limit: （演示）请求太频繁）"这样的提示。多问几次，看看连续失败时等待时间是不是在增长。把 DEMO_FAIL_RATE 设成 1，看看重试 4 次之后它怎么放弃。

实验 3：关掉流式，对比一下

在 cmd/mini-agent/main.go 里把 a.Streaming = true 改成 false，接上 DeepSeek，问同一个需要长回答的问题，对比两种模式下"等待的感觉"。总时间差不多，但你会明显感觉到非流式慢得多。这就是首 token 延迟和总延迟的区别。改完记得改回来。

实验 4：跑测试

go test ./internal/llm -run 'TestDeepSeekStream|TestRetryModel|TestDemoFailRateIsRetryable' -v
go test ./agent -run TestMaxTokensDuringToolUse -v

• TestDeepSeekStream：用 httptest 起一个假的服务端，返回 1.2 节那段 SSE，检查文字、思考、工具名的增量都交给了 handler，工具参数被正确拼接成 {"path":"a.go"}，用量也拿到了；
• TestDeepSeekStreamCutOffIsRetryable：流写到一半就断了，检查返回的错误是"可重试的"；
• TestRetryModel：先 503 再 429，第三次成功，检查重试了两次、第二次等待时间遵守了 retry-after；400 不重试；一直失败时最多重试指定次数；
• TestMaxTokensDuringToolUse：模型的输出在工具调用中被截断，检查工具没有执行、得到了一个提到 max_tokens 的错误结果、循环继续了。

五、踩坑记录：流式与重试的七个坑

坑 1：流断了，拿着半截结果当完整的用。
没收到 [DONE]、也没收到 finish_reason，就当成正常结束处理。文字少了一截还好，工具参数少了一截就可能执行出错误的操作。检查流是不是正常结束的。

坑 2：工具参数的增量没按 index 拼。
一次回复里有两个工具调用，参数片段交替到达。如果不按 index 区分，两个调用的参数就拼到一起了。

坑 3：对所有错误都重试。
400 重试五次，每次都失败，用户多等了一分钟。只重试"等一等就好"的错误。

坑 4：固定间隔重试，不加抖动。
服务端恢复的那一刻，所有客户端一起冲上来，又把它打挂了。指数退避加随机抖动，有 retry-after 就听它的。

坑 5：重试的等待不能被打断。
用户按了 Ctrl+C，程序还在 time.Sleep(30 * time.Second)。等待要监听 context。

坑 6：截断的工具调用被执行了，或者没给结果。
执行了，可能把文件写成半截；没给结果，下一次请求违反铁律被拒绝。不执行，但给一个说明原因的错误结果。

坑 7：把工具执行也包进自动重试。
"反正重试是个好东西，工具失败了也重试三次吧。"git push 超时了，其实已经推上去了，重试又推一次；一条插入数据库的命令，重试就是插入两条。有副作用的操作不能盲目重试。工具失败，交给看得见结果的模型去决定。

一句话总结

流式输出是 SSE：逐行解析 data，文字增量立刻显示，工具参数按 index 拼接，最后组装成和非流式一模一样的响应，流断了要识别出来；重试做成一个包装器，只重试限流、过载和网络错误，指数退避加随机抖动、听服务端的 retry-after；任务本身的错误交给模型；输出截断在工具调用中间时，不执行，但给一个结果。