Method runModelStepStreaming

pkg/agent/streaming.go:306–638

runModelStepStreaming 流式执行模型步骤 返回: (done, error)

(ctx context.Context, writer *stream.Writer[*session.Event])

// runModelStepStreaming 流式执行模型步骤
// 返回: (done, error)
func (a *Agent) runModelStepStreaming(ctx context.Context, writer *stream.Writer[*session.Event]) (bool, error) {
	// 1. 准备消息
	a.mu.RLock()
	messages := make([]types.Message, len(a.messages))
	copy(messages, a.messages)
	a.mu.RUnlock()

	// 2. 通过 Middleware 调用 LLM
	var resp *middleware.ModelResponse
	var err error

	streamLog.Debug(ctx, "using middleware stack", map[string]any{"has_stack": a.middlewareStack != nil})

	if a.middlewareStack != nil {
		// 使用 Middleware Stack
		streamLog.Debug(ctx, "using middleware stack for streaming", nil)
		// 转换工具列表
		toolList := make([]tools.Tool, 0, len(a.toolMap))
		for _, tool := range a.toolMap {
			toolList = append(toolList, tool)
		}

		req := &middleware.ModelRequest{
			Messages:     messages,
			SystemPrompt: a.template.SystemPrompt,
			Tools:        toolList,
			Metadata:     make(map[string]any),
		}

		// 创建适配器处理provider调用
		finalHandler := func(ctx context.Context, req *middleware.ModelRequest) (*middleware.ModelResponse, error) {
			// 转换工具定义
			toolSchemas := make([]provider.ToolSchema, len(req.Tools))
			for i, tool := range req.Tools {
				toolSchemas[i] = provider.ToolSchema{
					Name:        tool.Name(),
					Description: tool.Description(),
					InputSchema: tool.InputSchema(),
				}
			}

			// 创建Provider选项
			streamOpts := &provider.StreamOptions{
				Tools:       toolSchemas,
				System:      req.SystemPrompt,
				Temperature: 0.7,
			}

			// 调用Provider - 使用Stream方法支持流式响应
			streamLog.Debug(ctx, "calling provider.Stream() for middleware", nil)
			chunkCh, err := a.provider.Stream(ctx, req.Messages, streamOpts)
			if err != nil {
				return nil, err
			}

			// 直接将流式响应发送到WebSocket，但这里先收集用于兼容旧逻辑
			streamLog.Debug(ctx, "starting to collect chunks from provider", nil)
			var assistantMessage types.Message