Go: add stream / think chat (#14242)

### What problem does this PR solve?

1. Supports stream and non-stream chat
2. Supports think and non-think chat
3. List supported models from DeepSeek service. (This command can be
used to verify the API validity)

### Type of change

- [x] New Feature (non-breaking change which adds functionality)

---------

Signed-off-by: Jin Hai <haijin.chn@gmail.com>

internal/cli/client.go

@@ -164,6 +164,8 @@ func (c *RAGFlowClient) ExecuteAdminCommand(cmd *Command) (ResponseIf, error) {
 		return c.ShowProvider(cmd)
 	case "list_provider_models":
 		return c.ListModels(cmd)
+	case "list_supported_models":
+		return c.ListSupportedModels(cmd)
 	case "list_instance_models":
 		return c.ListInstanceModels(cmd)
 	case "show_model":
@@ -214,6 +216,8 @@ func (c *RAGFlowClient) ExecuteUserCommand(cmd *Command) (ResponseIf, error) {
 		return c.ShowProvider(cmd)
 	case "list_provider_models":
 		return c.ListModels(cmd)
+	case "list_supported_models":
+		return c.ListSupportedModels(cmd)
 	case "list_instance_models":
 		return c.ListInstanceModels(cmd)
 	case "show_model":

internal/cli/common_command.go

@@ -335,6 +335,45 @@ func (c *RAGFlowClient) ListModels(cmd *Command) (ResponseIf, error) {
 	return &result, nil
 }
 
+func (c *RAGFlowClient) ListSupportedModels(cmd *Command) (ResponseIf, error) {
+
+	providerName, ok := cmd.Params["provider_name"].(string)
+	if !ok {
+		return nil, fmt.Errorf("provider_name not provided")
+	}
+	instanceName, ok := cmd.Params["instance_name"].(string)
+	if !ok {
+		return nil, fmt.Errorf("instance_name not provided")
+	}
+
+	var endPoint string
+	if c.ServerType == "admin" {
+		endPoint = fmt.Sprintf("/admin/providers/%s/instances/%s/models?supported=true", providerName, instanceName)
+	} else {
+		endPoint = fmt.Sprintf("/providers/%s/instances/%s/models?supported=true", providerName, instanceName)
+	}
+
+	resp, err := c.HTTPClient.Request("GET", endPoint, true, "web", nil, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list models: %w", err)
+	}
+
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("failed to list models: HTTP %d, body: %s", resp.StatusCode, string(resp.Body))
+	}
+
+	var result CommonResponse
+	if err = json.Unmarshal(resp.Body, &result); err != nil {
+		return nil, fmt.Errorf("failed to list models: invalid JSON (%w)", err)
+	}
+
+	if result.Code != 0 {
+		return nil, fmt.Errorf("%s", result.Message)
+	}
+	result.Duration = resp.Duration
+	return &result, nil
+}
+
 func (c *RAGFlowClient) ShowModel(cmd *Command) (ResponseIf, error) {
 	providerName, ok := cmd.Params["provider_name"].(string)
 	if !ok {

internal/cli/lexer.go

@@ -303,6 +303,8 @@ func (l *Lexer) lookupIdent(ident string) Token {
 		return Token{Type: TokenChat, Value: ident}
 	case "THINK":
 		return Token{Type: TokenThink, Value: ident}
+	case "STREAM":
+		return Token{Type: TokenStream, Value: ident}
 	case "LS":
 		return Token{Type: TokenLS, Value: ident}
 	case "CAT":
@@ -363,6 +365,8 @@ func (l *Lexer) lookupIdent(ident string) Token {
 		return Token{Type: TokenTable, Value: ident}
 	case "AVAILABLE":
 		return Token{Type: TokenAvailable, Value: ident}
+	case "SUPPORTED":
+		return Token{Type: TokenSupported, Value: ident}
 	case "NAME":
 		return Token{Type: TokenName, Value: ident}
 	case "INSTANCE":

internal/cli/parser.go

@@ -190,6 +190,8 @@ func (p *Parser) parseUserCommand() (*Command, error) {
 		return p.parseEnableCommand()
 	case TokenDisable:
 		return p.parseDisableCommand()
+	case TokenStream:
+		return p.parseStreamCommand()
 	case TokenChat:
 		return p.parseChatCommand()
 	case TokenThink:

internal/cli/response.go

@@ -113,28 +113,33 @@ func (r *SimpleResponse) PrintOut() {
 	}
 }
 
-type MessageResponse struct {
-	Code         int    `json:"code"`
-	Message      string `json:"message"`
-	Duration     float64
-	OutputFormat OutputFormat
+type NonStreamResponse struct {
+	Code             int    `json:"code"`
+	ReasoningContent string `json:"reasoning_content"`
+	Answer           string `json:"answer"`
+	Message          string `json:"message"`
+	Duration         float64
+	OutputFormat     OutputFormat
 }
 
-func (r *MessageResponse) Type() string {
-	return "message"
+func (r *NonStreamResponse) Type() string {
+	return "non_stream_message"
 }
 
-func (r *MessageResponse) TimeCost() float64 {
+func (r *NonStreamResponse) TimeCost() float64 {
 	return r.Duration
 }
 
-func (r *MessageResponse) SetOutputFormat(format OutputFormat) {
+func (r *NonStreamResponse) SetOutputFormat(format OutputFormat) {
 	r.OutputFormat = format
 }
 
-func (r *MessageResponse) PrintOut() {
+func (r *NonStreamResponse) PrintOut() {
 	if r.Code == 0 {
-		fmt.Println(r.Message)
+		if r.ReasoningContent != "" {
+			fmt.Printf("Thinking: %s\n", r.ReasoningContent)
+		}
+		fmt.Printf("Answer: %s\n", r.Answer)
 	} else {
 		fmt.Println("ERROR")
 		fmt.Printf("%d, %s\n", r.Code, r.Message)

internal/cli/types.go

@@ -73,6 +73,7 @@ const (
 	TokenKeys
 	TokenGenerate
 	TokenAvailable
+	TokenSupported
 	TokenModel
 	TokenModels
 	TokenProvider
@@ -80,6 +81,7 @@ const (
 	TokenDefault
 	TokenChats
 	TokenChat
+	TokenStream
 	TokenFiles
 	TokenAs
 	TokenParse
@@ -106,7 +108,6 @@ const (
 	TokenIndex
 	TokenVector
 	TokenSize
-	TokenDocMeta
 	TokenName // For ALTER PROVIDER <name> NAME <new_name>
 	TokenInstance
 	TokenInstances

internal/cli/user_command.go

@@ -1436,83 +1436,106 @@ func (c *RAGFlowClient) ChatToModel(cmd *Command) (ResponseIf, error) {
 	}
 
 	message := cmd.Params["message"].(string)
-	reasoning := cmd.Params["reasoning"].(bool)
+	thinking := cmd.Params["thinking"].(bool)
+	stream := cmd.Params["stream"].(bool)
 
 	url := fmt.Sprintf("/providers/%s/instances/%s/models/%s", providerName, instanceName, modelName)
 
 	payload := map[string]interface{}{
-		"message":   message,
-		"stream":    true, // use stream API
-		"reasoning": reasoning,
+		"message":  message,
+		"stream":   stream, // use stream API
+		"thinking": thinking,
 	}
 
-	// Call stream http api
-	reader, duration, err := c.HTTPClient.RequestStream("POST", url, true, "web", nil, payload)
-	if err != nil {
-		return nil, fmt.Errorf("failed to chat model: %w", err)
-	}
-	defer reader.Close()
-
-	// Parse SSE and output to console
-	scanner := bufio.NewScanner(reader)
-	var fullMessage strings.Builder
-
-	reasoningPrint := true
-	messagePrint := true
-	for scanner.Scan() {
-		line := scanner.Text()
-		if strings.HasPrefix(line, "data:") {
-			data := strings.TrimPrefix(line, "data:")
-			data = strings.TrimSpace(data)
-
-			if strings.HasPrefix(data, "[REASONING]") {
-				data = strings.TrimPrefix(data, "[REASONING]")
-				if reasoningPrint {
-					fmt.Print("Thinking: ")
-					reasoningPrint = false
-				} else {
-					fmt.Print(data)
-				}
-				os.Stdout.Sync()
-			}
-			if strings.HasPrefix(data, "[MESSAGE]") {
-				data = strings.TrimPrefix(data, "[MESSAGE]")
-				if messagePrint {
-					if reasoning {
-						fmt.Println()
-					}
-					fmt.Print("Answer: ")
-					messagePrint = false
-				} else {
-					fmt.Print(data)
-					os.Stdout.Sync()
-					fullMessage.WriteString(data)
-				}
-			}
-		} else if strings.HasPrefix(line, "event:error") {
-			// error event
-			if scanner.Scan() {
-				errData := strings.TrimPrefix(scanner.Text(), "data:")
-				errData = strings.TrimSpace(errData)
-				return nil, fmt.Errorf("chat error: %s", errData)
-			}
-			// If there's an error, return a generic error
-			return nil, fmt.Errorf("chat error: received error event from server")
+	if stream {
+		// Call stream http api
+		reader, duration, err := c.HTTPClient.RequestStream("POST", url, true, "web", nil, payload)
+		if err != nil {
+			return nil, fmt.Errorf("failed to chat model: %w", err)
 		}
+		defer reader.Close()
+
+		// Parse SSE and output to console
+		scanner := bufio.NewScanner(reader)
+		var fullMessage strings.Builder
+
+		reasoningPrint := true
+		messagePrint := true
+		for scanner.Scan() {
+			line := scanner.Text()
+			if strings.HasPrefix(line, "data:") {
+				data := strings.TrimPrefix(line, "data:")
+				data = strings.TrimSpace(data)
+
+				if strings.HasPrefix(data, "[REASONING]") {
+					data = strings.TrimPrefix(data, "[REASONING]")
+					if reasoningPrint {
+						fmt.Print("Thinking: ")
+						reasoningPrint = false
+					} else {
+						fmt.Print(data)
+					}
+					os.Stdout.Sync()
+				}
+				if strings.HasPrefix(data, "[MESSAGE]") {
+					data = strings.TrimPrefix(data, "[MESSAGE]")
+					if messagePrint {
+						if thinking {
+							fmt.Println()
+						}
+						fmt.Print("Answer: ")
+						messagePrint = false
+					} else {
+						fmt.Print(data)
+						os.Stdout.Sync()
+						fullMessage.WriteString(data)
+					}
+				}
+			} else if strings.HasPrefix(line, "event:error") {
+				// error event
+				if scanner.Scan() {
+					errData := strings.TrimPrefix(scanner.Text(), "data:")
+					errData = strings.TrimSpace(errData)
+					return nil, fmt.Errorf("chat error: %s", errData)
+				}
+				// If there's an error, return a generic error
+				return nil, fmt.Errorf("chat error: received error event from server")
+			}
+		}
+
+		if err := scanner.Err(); err != nil {
+			return nil, fmt.Errorf("error reading stream: %w", err)
+		}
+
+		fmt.Println()
+
+		result := &StreamMessageResponse{
+			Code:     0,
+			Message:  fullMessage.String(),
+			Duration: duration,
+		}
+		return result, nil
 	}
 
-	if err := scanner.Err(); err != nil {
-		return nil, fmt.Errorf("error reading stream: %w", err)
+	resp, err := c.HTTPClient.Request("POST", url, true, "web", nil, payload)
+	if err != nil {
+		return nil, fmt.Errorf("failed to list instance models: %w", err)
 	}
 
-	fmt.Println()
-
-	result := &StreamMessageResponse{
-		Code:     0,
-		Message:  fullMessage.String(),
-		Duration: duration,
+	if resp.StatusCode != 200 {
+		return nil, fmt.Errorf("failed to list instance models: HTTP %d, body: %s", resp.StatusCode, string(resp.Body))
 	}
-	return result, nil
+
+	var result NonStreamResponse
+	if err = json.Unmarshal(resp.Body, &result); err != nil {
+		return nil, fmt.Errorf("failed to list instance models: invalid JSON (%w)", err)
+	}
+
+	if result.Code != 0 {
+		return nil, fmt.Errorf("%s", result.Message)
+	}
+	result.Duration = resp.Duration
+	return &result, nil
 }
 
 // UseModel sets the current model for chat

internal/cli/user_parser.go

@@ -163,6 +163,8 @@ func (p *Parser) parseListCommand() (*Command, error) {
 		return p.parseListTokens()
 	case TokenModel:
 		return p.parseListModelProviders()
+	case TokenSupported:
+		return p.parseListModelsOfProvider()
 	case TokenModels:
 		return p.parseListModelsOfProvider()
 	case TokenProviders:
@@ -2014,11 +2016,55 @@ func (p *Parser) parseSearchCommand() (*Command, error) {
 }
 
 func (p *Parser) parseListModelsOfProvider() (*Command, error) {
+
+	if p.curToken.Type == TokenSupported {
+		// List supported models
+		p.nextToken()
+
+		cmd := NewCommand("list_supported_models")
+		if p.curToken.Type != TokenModels {
+			return nil, fmt.Errorf("expected MODELS")
+		}
+		p.nextToken()
+
+		if p.curToken.Type != TokenFrom {
+			return nil, fmt.Errorf("expected FROM")
+		}
+		p.nextToken()
+
+		if p.curToken.Type != TokenQuotedString {
+			return nil, fmt.Errorf("expected quoted string for provider name")
+		}
+		firstName, err := p.parseQuotedString()
+		if err != nil {
+			return nil, err
+		}
+		p.nextToken()
+
+		if p.curToken.Type != TokenQuotedString {
+			return nil, fmt.Errorf("expected quoted string for instance name")
+		}
+		secondName, err := p.parseQuotedString()
+		if err != nil {
+			return nil, err
+		}
+		p.nextToken()
+
+		cmd.Params["provider_name"] = firstName
+		cmd.Params["instance_name"] = secondName
+
+		// Semicolon is optional for UNSET TOKEN
+		if p.curToken.Type == TokenSemicolon {
+			p.nextToken()
+		}
+		return cmd, nil
+	}
+
 	if p.curToken.Type != TokenModels {
 		return nil, fmt.Errorf("expected MODELS")
 	}
-
 	p.nextToken()
+
 	if p.curToken.Type != TokenFrom {
 		return nil, fmt.Errorf("expected FROM")
 	}
@@ -2194,19 +2240,47 @@ func (p *Parser) parseChatCommand() (*Command, error) {
 		cmd.Params["composite_model_name"] = compositeModelName
 	}
 	cmd.Params["message"] = message
-	cmd.Params["reasoning"] = false
+	cmd.Params["thinking"] = false
+	cmd.Params["stream"] = false
 	return cmd, nil
 }
 
 func (p *Parser) parseThinkCommand() (*Command, error) {
 
 	p.nextToken() // consume THINK
+
+	if p.curToken.Type != TokenChat {
+		return nil, fmt.Errorf("expected CHAT after THINK")
+	}
+
 	command, err := p.parseChatCommand()
 	if err != nil {
 		return nil, err
 	}
-	command.Type = "think_chat_to_model"
-	command.Params["reasoning"] = true
+	command.Params["thinking"] = true
+	return command, nil
+}
+
+func (p *Parser) parseStreamCommand() (*Command, error) {
+
+	p.nextToken() // consume STREAM
+
+	var command *Command
+	var err error
+
+	if p.curToken.Type == TokenChat {
+		command, err = p.parseChatCommand()
+		if err != nil {
+			return nil, err
+		}
+	} else if p.curToken.Type == TokenThink {
+		command, err = p.parseThinkCommand()
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	command.Params["stream"] = true
 	return command, nil
 }