从阻塞到事件流：流式输出怎么做

用户提问 → [模型开始想...] → [决定调工具] → [工具结果回来了]
         → [模型继续想...] → [第一个字] → [第二个字] → ... → [最终答案]

生产者（Agent 循环）              消费者（前端/调用方）
    │                                  │
    │  产出事件                         │
    ├──────────► channel ◄─────────────┤
    │           (缓冲队列)              │  消费事件
    │                                  │

// agents/stream.go:45
type StreamedResult struct {
    ch chan streamMsg  // 事件 channel

    mu    sync.Mutex
    final *RunResult   // 循环结束后的最终结果
    err   error
}

type streamMsg struct {
    event StreamEvent  // 事件（可能为 nil）
    err   error        // 终止错误（可能为 nil）
}

// agents/stream.go:61
func (s *StreamedResult) Events() iter.Seq2[StreamEvent, error] {
    return func(yield func(StreamEvent, error) bool) {
        for msg := range s.ch {  // 迭代 channel
            if msg.err != nil {
                yield(nil, msg.err)  // 终止错误
                return
            }
            if !yield(msg.event, nil) {
                return  // 消费者提前 break
            }
        }
    }
}

// agents/stream.go:77
func (s *StreamedResult) FinalResult() (*RunResult, error) {
    // 迭代完 channel 后调用，拿最终结果
}

sr := agents.RunStreamed(ctx, agent, "帮我查天气", opts)

// 1. 消费事件流
for event, err := range sr.Events() {
    if err != nil { panic(err) }
    switch e := event.(type) {
    case *agents.RunItemStreamEvent:
        if e.Name == "message_output_created" {
            fmt.Print(e.Item.(*agents.MessageOutputItem).Text())
        }
    }
}

// 2. 拿最终结果
res, _ := sr.FinalResult()

// agents/stream.go:94
func RunStreamed(ctx context.Context, agent *Agent, input any, opts RunOptions) *StreamedResult {
    sr := &StreamedResult{ch: make(chan streamMsg, 64)}

    go func() {  // 起一个后台 goroutine 跑循环
        defer close(sr.ch)  // 循环结束，关闭 channel（消费者迭代会自动结束）
        res, err := runStreamedLoop(ctx, agent, input, opts, sr)
        sr.setFinal(res, err)  // 存最终结果
        if err != nil {
            sr.ch <- streamMsg{err: err}  // 发终止错误
        }
    }()

    return sr  // 立即返回，不阻塞
}

ch: make(chan streamMsg, 64)

// agents/stream.go:111
func (s *StreamedResult) emit(ctx context.Context, event StreamEvent) {
    select {
    case s.ch <- streamMsg{event: event}:  // 正常发送
    case <-ctx.Done():                      // context 取消了，丢弃事件
    }
}

// agents/stream.go:118
switch item.(type) {
case *MessageOutputItem:    return "message_output_created"
case *ToolCallItem:         return "tool_called"
case *ToolCallOutputItem:   return "tool_output"
case *HandoffCallItem:      return "handoff_requested"
case *HandoffOutputItem:    return "handoff_occured"
case *ReasoningItem:        return "reasoning_item_created"
}

// agents/stream_run.go:301（骨架）
func streamOneModelCall(...) iter.Seq2[*TResponseStreamEvent, error] {
    return func(yield func(*TResponseStreamEvent, error) bool) {
        for event, err := range model.StreamResponse(ctx, req) {
            if err != nil { yield(nil, err); return }

            // 透传原始事件（消费者能拿到 response.output_text.delta）
            if !yield(event, nil) { return }

            // 收集 response.completed 事件，组装最终响应
            if event.Type == "response.completed" {
                finalResponse = assembleFrom(event)
            }
        }
        // 没有 completed 事件 → 报错（不伪造空响应）
    }
}

// 阻塞版 run.go:299
go func() { guardResult := runInputGuardrails(...) }()  // 后台
resp := model.GetResponse(...)  // 同时调模型

// 流式版 stream_run.go:140
guardResult := runInputGuardrails(...)  // 先跑完
streamOneModelCall(...)  // 再开始流式

ctx, cancel := context.WithCancel(parentCtx)
defer cancel()  // 确保最终取消

sr := agents.RunStreamed(ctx, agent, input, opts)

for event, err := range sr.Events() {
    if someCondition {
        break  // 提前退出
    }
}
// cancel() 会通过 defer 触发，生产者感知 ctx.Done() 后退出

// mini-claude-code/model.go 的 ChatStream
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
    line := scanner.Text()              // 逐行读 SSE
    if !strings.HasPrefix(line, "data: ") { continue }
    payload := strings.TrimPrefix(line, "data: ")
    if payload == "[DONE]" { break }

    var chunk struct{ Choices []... }
    json.Unmarshal([]byte(payload), &chunk)

    delta := chunk.Choices[0].Delta
    if delta.ReasoningContent != "" {   // 思考过程(glm-5.2)
        onEvent(StreamEvent{ReasonDelta: delta.ReasoningContent})
    }
    if delta.Content != "" {            // 正文
        onEvent(StreamEvent{TextDelta: delta.Content})
    }
    // 工具调用按 index 累积参数分片...
}