Method processStream

pkg/provider/glm.go:362–414

processStream 处理流式响应

(body io.ReadCloser, chunkCh chan<- StreamChunk)

// processStream 处理流式响应
func (gp *GLMProvider) processStream(body io.ReadCloser, chunkCh chan<- StreamChunk) {
	defer close(chunkCh)
	defer func() { _ = body.Close() }()

	ctx := context.Background() // 用于日志记录

	scanner := bufio.NewScanner(body)
	eventCount := 0
	for scanner.Scan() {
		line := scanner.Text()

		// SSE格式: "data: {...}"
		if !strings.HasPrefix(line, "data: ") {
			// 记录非数据行（用于调试）
			if strings.TrimSpace(line) != "" && !strings.HasPrefix(line, ":") {
				glmLog.Debug(ctx, "non-data line", map[string]any{"line": line})
			}
			continue
		}

		data := strings.TrimPrefix(line, "data: ")

		// 忽略特殊标记
		if data == "[DONE]" {
			glmLog.Debug(ctx, "received [DONE] marker", nil)
			break
		}

		// 解析JSON
		var event map[string]any
		if err := json.Unmarshal([]byte(data), &event); err != nil {
			glmLog.Debug(ctx, "failed to parse JSON", map[string]any{"error": err, "data": data})
			continue
		}

		eventCount++
		glmLog.Debug(ctx, "stream event", map[string]any{"event_num": eventCount, "event": event})

		chunk := gp.parseStreamEvent(event)
		if chunk != nil {
			glmLog.Debug(ctx, "parsed chunk", map[string]any{"type": chunk.Type, "index": chunk.Index})
			chunkCh <- *chunk
		} else {
			glmLog.Debug(ctx, "no chunk parsed from event", nil)
		}
	}

	if err := scanner.Err(); err != nil {
		glmLog.Error(ctx, "scanner error", map[string]any{"error": err})
	}

	glmLog.Debug(ctx, "processed events", map[string]any{"total": eventCount})
}

// parseStreamEvent 解析流式事件
func (gp *GLMProvider) parseStreamEvent(event map[string]any) *StreamChunk {