feat: add fluent workflow

This is equivalent to #228 but
with the data extraction parts removed. We will revisit that in a
separate PR.

Author: rcrisanti

VERSION

@@ -1 +1 @@
-v1.1.0
+v1.2.2

main.go

@@ -24,6 +24,7 @@ package main
 
 import (
 	_ "embed"
+	"strings"
 
 	"github.com/ansys/aali-sharedtypes/pkg/config"
 	"github.com/ansys/aali-sharedtypes/pkg/logging"
@@ -33,6 +34,9 @@ import (
 	"github.com/ansys/aali-flowkit/pkg/internalstates"
 )
 
+//go:embed VERSION
+var version string
+
 //go:embed pkg/externalfunctions/dataextraction.go
 var dataExtractionFile string
 
@@ -69,6 +73,9 @@ var mcpFile string
 //go:embed pkg/externalfunctions/rhsc.go
 var rhscFile string
 
+//go:embed pkg/externalfunctions/fluent.go
+var fluentFile string
+
 func init() {
 	// initialize config
 	config.InitConfig([]string{}, map[string]interface{}{
@@ -83,6 +90,9 @@ func init() {
 
 	// initialize logging
 	logging.InitLogger(config.GlobalConfig)
+
+	// assign the version from the embedded file
+	internalstates.FlowkitVersion = strings.TrimSpace(version)
 }
 
 func main() {
@@ -103,6 +113,7 @@ func main() {
 		"auth":             authFile,
 		"mcp":              mcpFile,
 		"rhsc":             rhscFile,
+		"fluent":           fluentFile,
 	}
 
 	// Load function definitions

pkg/externalfunctions/ansysgpt.go

@@ -24,9 +24,12 @@ package externalfunctions
 
 import (
 	"bytes"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"io"
 	"net/http"
+	"net/url"
 	"regexp"
 	"sort"
 	"strconv"
@@ -897,6 +900,99 @@ func AecGetContextFromRetrieverModule(
 	return context
 }
 
+// GetContextFromDataPlugin retrieves context from a data plugin
+//
+// Tags:
+//   - @displayName: Get Context from Data Plugin
+//
+// Parameters:
+//   - userQuery: the user query
+//   - apiUrl: the API URL of the data plugin
+//   - username: the username for authentication at the data plugin
+//   - password: the password for authentication at the data plugin
+//   - topK: the number of results to be returned
+//
+// Returns:
+//   - context: the context retrieved from the data plugin
+func GetContextFromDataPlugin(userQuery string, apiUrl string, username string, password string, topK int) (context []sharedtypes.AnsysGPTRetrieverModuleChunk) {
+	// Encode the query and max_doc in base64
+	encodedQuery := base64.StdEncoding.EncodeToString([]byte(userQuery))
+	encodedMaxDoc := base64.StdEncoding.EncodeToString([]byte(strconv.Itoa(topK)))
+
+	// Prepare form data
+	formData := url.Values{}
+	formData.Set("q", encodedQuery)
+	formData.Set("max_doc", encodedMaxDoc)
+
+	// Create the request
+	req, err := http.NewRequest("POST", apiUrl, bytes.NewBufferString(formData.Encode()))
+	if err != nil {
+		panic(fmt.Errorf("error creating request: %v", err))
+	}
+
+	// Set content type for form data
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+
+	// Set basic authentication header
+	auth := username + ":" + password
+	encodedAuth := base64.StdEncoding.EncodeToString([]byte(auth))
+	req.Header.Set("Authorization", "Basic "+encodedAuth)
+
+	// Make the request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		panic(fmt.Errorf("error making request: %v", err))
+	}
+	defer resp.Body.Close()
+
+	// Read the response body
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		panic(fmt.Errorf("error reading response body: %v", err))
+	}
+
+	// Check status code
+	if resp.StatusCode != 200 {
+		panic(fmt.Errorf("error response from data plugin: %v, body: %s", resp.Status, string(body)))
+	}
+
+	// The response is base64 encoded
+	base64EncodedResponse := string(body)
+
+	// Decode the base64 response
+	decodedResponse, err := base64.StdEncoding.DecodeString(base64EncodedResponse)
+	if err != nil {
+		panic(fmt.Errorf("error decoding base64 response: %v", err))
+	}
+
+	// Unmarshal the response
+	response := map[string]sharedtypes.AnsysGPTRetrieverModuleChunk{}
+	err = json.Unmarshal(decodedResponse, &response)
+	if err != nil {
+		panic(fmt.Errorf("error unmarshalling response: %v", err))
+	}
+	logging.Log.Debugf(&logging.ContextMap{}, "Received response from retriever module: %v", response)
+
+	// Extract the context from the response
+	context = make([]sharedtypes.AnsysGPTRetrieverModuleChunk, len(response))
+	for chunkNum, chunk := range response {
+		// Extract int from chunkNum
+		_, chunkNumstring, found := strings.Cut(chunkNum, "chunk ")
+		if !found {
+			panic(fmt.Errorf("error extracting chunk number from '%v'", chunkNum))
+		}
+		chunkNumInt, err := strconv.Atoi(chunkNumstring)
+		if err != nil {
+			panic(fmt.Errorf("error converting chunk number to int: %v", err))
+		}
+		// Store the chunk in the context slice
+		context[chunkNumInt-1] = chunk
+	}
+
+	return context
+}
+
 // AecPerformLLMFinalRequest performs a final request to LLM
 //
 // Tags:
@@ -927,7 +1023,8 @@ func AecPerformLLMFinalRequest(systemTemplate string,
 	tokenCountModelName string,
 	isStream bool,
 	userEmail string,
-	jwtToken string) (message string, stream *chan string) {
+	jwtToken string,
+	dontSendTokenCount bool) (message string, stream *chan string) {
 
 	logging.Log.Debugf(&logging.ContextMap{}, "Performing LLM final request")
 
@@ -1018,8 +1115,14 @@ func AecPerformLLMFinalRequest(systemTemplate string,
 	}
 	totalInputTokenCount := previousInputTokenCount + inputTokenCount
 
+	// check if token count should be sent
+	sendTokenCount := false
+	if !dontSendTokenCount {
+		sendTokenCount = true
+	}
+
 	// Start a goroutine to transfer the data from the response channel to the stream channel.
-	go transferDatafromResponseToStreamChannel(&responseChannel, &streamChannel, false, true, tokenCountEndpoint, totalInputTokenCount, previousOutputTokenCount, tokenCountModelName, jwtToken, userEmail, true, contextString)
+	go transferDatafromResponseToStreamChannel(&responseChannel, &streamChannel, false, sendTokenCount, tokenCountEndpoint, totalInputTokenCount, previousOutputTokenCount, tokenCountModelName, jwtToken, userEmail, true, contextString)
 
 	return "", &streamChannel
 }

pkg/externalfunctions/externalfunctions.go

@@ -78,6 +78,7 @@ var ExternalFunctionsMap = map[string]interface{}{
 	"AisChangeAcsResponsesByFactor":                  AisChangeAcsResponsesByFactor,
 	"AecGetContextFromRetrieverModule":               AecGetContextFromRetrieverModule,
 	"AecPerformLLMFinalRequest":                      AecPerformLLMFinalRequest,
+	"GetContextFromDataPlugin":                       GetContextFromDataPlugin,
 
 	// data extraction
 	"GetGithubFilesToExtract":                    GetGithubFilesToExtract,
@@ -101,6 +102,7 @@ var ExternalFunctionsMap = map[string]interface{}{
 	"JsonPath":               JsonPath,
 	"StringConcat":           StringConcat,
 	"StringFormat":           StringFormat,
+	"FluentCodeGenTest":      FluentCodeGenTest,
 
 	// code generation
 	"LoadCodeGenerationElements":      LoadCodeGenerationElements,
@@ -187,4 +189,7 @@ var ExternalFunctionsMap = map[string]interface{}{
 
 	// rhsc
 	"SetCopilotGenerateRequestJsonBody": SetCopilotGenerateRequestJsonBody,
+
+	// fluent
+	"FluentCodeGen": FluentCodeGen,
 }

pkg/externalfunctions/fluent.go

@@ -0,0 +1,105 @@
+// Copyright (C) 2025 ANSYS, Inc. and/or its affiliates.
+// SPDX-License-Identifier: MIT
+//
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in all
+// copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+// SOFTWARE.
+
+package externalfunctions
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"strings"
+)
+
+// FluentCodeGen sends a raw user message to the Fluent container and returns the response
+//
+// Tags:
+//   - @displayName: Fluent Code Gen
+//
+// Parameters:
+//   - message: the raw user message to send to the container
+//
+// Returns:
+//   - response: the response from the Fluent container as a string
+func FluentCodeGen(message string) (response string) {
+	url := "http://aali-fluent:8000/chat"
+
+	// Create the JSON payload directly
+	jsonData := fmt.Sprintf(`{"message": "%s"}`, message)
+
+	// Create HTTP request
+	req, err := http.NewRequest("POST", url, bytes.NewBufferString(jsonData))
+	if err != nil {
+		panic(fmt.Sprintf("Error creating HTTP request: %v", err))
+	}
+
+	// Set headers
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Content-Type", "application/json")
+
+	// Execute the request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		panic(fmt.Sprintf("Error executing HTTP request: %v", err))
+	}
+	defer resp.Body.Close()
+
+	// Read the response body
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		panic(fmt.Sprintf("Error reading response body: %v", err))
+	}
+
+	// Check if the response code is successful (2xx)
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		panic(fmt.Sprintf("HTTP request failed with status code %d: %s", resp.StatusCode, string(body)))
+	}
+
+	// Parse JSON response to extract just the response content
+	var responseData map[string]interface{}
+	if err := json.Unmarshal(body, &responseData); err != nil {
+		panic(fmt.Sprintf("Error parsing JSON response: %v", err))
+	}
+
+	// Extract the response field
+	if responseField, exists := responseData["response"]; exists {
+		if responseArray, ok := responseField.([]interface{}); ok && len(responseArray) > 0 {
+			// Concatenate all items in the response array with a newline
+			var concatenatedResponse string
+			for _, item := range responseArray {
+				if str, ok := item.(string); ok {
+					concatenatedResponse += str + "\n"
+				} else {
+					concatenatedResponse += fmt.Sprintf("%v\n", item)
+				}
+			}
+			// Remove the trailing newline
+			concatenatedResponse = strings.TrimRight(concatenatedResponse, "\n")
+			return "```python\n" + concatenatedResponse + "```"
+		}
+	}
+
+	// Fallback to raw response if parsing fails
+	return string(body)
+}

pkg/externalfunctions/generic.go

@@ -270,3 +270,77 @@ func StringFormat(data any, format string) string {
 	}
 	return fmt.Sprintf(format, data)
 }
+
+// FluentCodeGen sends a raw user message to the Fluent container and returns the response.
+// It takes a user message and posts it to the Fluent API endpoint to generate code.
+//
+// Tags:
+//   - @displayName: Fluent Code Gen Test
+//
+// Parameters:
+//   - message: the raw user message to send to the container
+//
+// Returns:
+//   - response: the response from the Fluent container as a string
+func FluentCodeGenTest(message string) (response string) {
+	url := "http://aali-fluent:8000/chat"
+
+	// Create the JSON payload directly
+	jsonData := fmt.Sprintf(`{"message": "%s"}`, message)
+
+	// Create HTTP request
+	req, err := http.NewRequest("POST", url, bytes.NewBufferString(jsonData))
+	if err != nil {
+		panic(fmt.Sprintf("Error creating HTTP request: %v", err))
+	}
+
+	// Set headers
+	req.Header.Set("Accept", "application/json")
+	req.Header.Set("Content-Type", "application/json")
+
+	// Execute the request
+	client := &http.Client{}
+	resp, err := client.Do(req)
+	if err != nil {
+		panic(fmt.Sprintf("Error executing HTTP request: %v", err))
+	}
+	defer resp.Body.Close()
+
+	// Read the response body
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		panic(fmt.Sprintf("Error reading response body: %v", err))
+	}
+
+	// Check if the response code is successful (2xx)
+	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
+		panic(fmt.Sprintf("HTTP request failed with status code %d: %s", resp.StatusCode, string(body)))
+	}
+
+	// Parse JSON response to extract just the response content
+	var responseData map[string]interface{}
+	if err := json.Unmarshal(body, &responseData); err != nil {
+		panic(fmt.Sprintf("Error parsing JSON response: %v", err))
+	}
+
+	// Extract the response field
+	if responseField, exists := responseData["response"]; exists {
+		if responseArray, ok := responseField.([]interface{}); ok && len(responseArray) > 0 {
+			// Concatenate all items in the response array with a newline
+			var concatenatedResponse string
+			for _, item := range responseArray {
+				if str, ok := item.(string); ok {
+					concatenatedResponse += str + "\n"
+				} else {
+					concatenatedResponse += fmt.Sprintf("%v\n", item)
+				}
+			}
+			// Remove the trailing newline
+			concatenatedResponse = strings.TrimRight(concatenatedResponse, "\n")
+			return "```python\n" + concatenatedResponse + "```"
+		}
+	}
+
+	// Fallback to raw response if parsing fails
+	return string(body)
+}