Dify on wechat

chat/Dockerfile

@@ -1,4 +1,4 @@
-FROM golang:1.23.4-alpine AS builder
+FROM golang:1.24.1-alpine AS builder
 
 # 为镜像设置必要的环境变量
 ENV GO111MODULE=on \
@@ -40,6 +40,7 @@ COPY --from=builder /usr/local/bin/ffmpeg /bin/ffmpeg
 COPY --from=builder /usr/share/zoneinfo/Asia/Shanghai /usr/share/zoneinfo/Asia/Shanghai
 COPY --from=builder /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
 COPY --from=builder /build/service/chat/api/etc /etc
+COPY --from=builder /tmp /tmp
 ENV TZ Asia/Shanghai
 
 # 从builder镜像中把/build/app 拷贝到当前目录

chat/Dockerfile-websocket

@@ -1,4 +1,4 @@
-FROM golang:1.23.4-alpine AS builder
+FROM golang:1.24.1-alpine AS builder
 
 # 为镜像设置必要的环境变量
 ENV GO111MODULE=on \

chat/common/dify/api.go

@@ -0,0 +1,47 @@
+package dify
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+)
+
+type API struct {
+	c      *Client
+	secret string
+}
+
+func (api *API) WithSecret(secret string) *API {
+	api.secret = secret
+	return api
+}
+
+func (api *API) getSecret() string {
+	if api.secret != "" {
+		return api.secret
+	}
+	return api.c.getAPISecret()
+}
+
+func (api *API) createBaseRequest(ctx context.Context, method, apiUrl string, body interface{}) (*http.Request, error) {
+	var b io.Reader
+	if body != nil {
+		reqBytes, err := json.Marshal(body)
+		if err != nil {
+			return nil, err
+		}
+		b = bytes.NewBuffer(reqBytes)
+	} else {
+		b = http.NoBody
+	}
+	req, err := http.NewRequestWithContext(ctx, method, api.c.getHost()+apiUrl, b)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Authorization", "Bearer "+api.getSecret())
+	req.Header.Set("Cache-Control", "no-cache")
+	req.Header.Set("Content-Type", "application/json; charset=utf-8")
+	return req, nil
+}

chat/common/dify/api_audio_to_text.go

@@ -0,0 +1,51 @@
+package dify
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+
+	"github.com/go-resty/resty/v2"
+)
+
+// AudioToTextResponse represents the response from the audio-to-text API
+type AudioToTextResponse struct {
+	Text string `json:"text"`
+}
+
+// ErrorResponse represents the error response from the API
+type ErrorResponse struct {
+	Code    string `json:"code"`
+	Message string `json:"message"`
+	Status  int    `json:"status"`
+}
+
+func (api *API) AudioToText(ctx context.Context, filePath string) (string, error) {
+	client := resty.New()
+
+	resp, err := client.R().
+		SetContext(ctx).
+		SetHeader("Authorization", "Bearer "+api.getSecret()).
+		SetHeader("Cache-Control", "no-cache").
+		SetFile("file", filePath).
+		Post(api.c.getHost() + "/v1/audio-to-text")
+
+	if err != nil {
+		return "", fmt.Errorf("send request: %w", err)
+	}
+
+	if resp.StatusCode() != 200 {
+		var errResp ErrorResponse
+		if err := json.Unmarshal(resp.Body(), &errResp); err != nil {
+			return "", fmt.Errorf("HTTP error %d: failed to decode error response", resp.StatusCode())
+		}
+		return "", fmt.Errorf("API error: [%s] %s", errResp.Code, errResp.Message)
+	}
+
+	var result AudioToTextResponse
+	if err := json.Unmarshal(resp.Body(), &result); err != nil {
+		return "", fmt.Errorf("decode response: %w", err)
+	}
+
+	return result.Text, nil
+}

chat/common/dify/api_chat.go

@@ -0,0 +1,33 @@
+package dify
+
+import (
+	"context"
+	"net/http"
+)
+
+type ChatMessageRequest struct {
+	Inputs         map[string]interface{} `json:"inputs"`
+	Query          string                 `json:"query"`
+	ResponseMode   string                 `json:"response_mode"`
+	ConversationID string                 `json:"conversation_id,omitempty"`
+	User           string                 `json:"user"`
+}
+
+type ChatMessageResponse struct {
+	ID             string `json:"id"`
+	Answer         string `json:"answer"`
+	ConversationID string `json:"conversation_id"`
+	CreatedAt      int    `json:"created_at"`
+}
+
+// ChatMessages /* Create chat message
+func (api *API) ChatMessages(ctx context.Context, req *ChatMessageRequest) (resp *ChatMessageResponse, err error) {
+	req.ResponseMode = "blocking"
+
+	httpReq, err := api.createBaseRequest(ctx, http.MethodPost, "/v1/chat-messages", req)
+	if err != nil {
+		return
+	}
+	err = api.c.sendJSONRequest(httpReq, &resp)
+	return
+}

chat/common/dify/api_chat_stream.go

@@ -0,0 +1,95 @@
+package dify
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+)
+
+type ChatMessageStreamResponse struct {
+	Event          string `json:"event"`
+	TaskID         string `json:"task_id"`
+	ID             string `json:"id"`
+	Answer         string `json:"answer"`
+	CreatedAt      int64  `json:"created_at"`
+	ConversationID string `json:"conversation_id"`
+}
+
+type ChatMessageStreamChannelResponse struct {
+	ChatMessageStreamResponse
+	Err error `json:"-"`
+}
+
+func (api *API) ChatMessagesStreamRaw(ctx context.Context, req *ChatMessageRequest) (*http.Response, error) {
+	req.ResponseMode = "streaming"
+
+	httpReq, err := api.createBaseRequest(ctx, http.MethodPost, "/v1/chat-messages", req)
+	if err != nil {
+		return nil, err
+	}
+	return api.c.sendRequest(httpReq)
+}
+
+func (api *API) ChatMessagesStream(ctx context.Context, req *ChatMessageRequest) (chan ChatMessageStreamChannelResponse, error) {
+	httpResp, err := api.ChatMessagesStreamRaw(ctx, req)
+	if err != nil {
+		return nil, err
+	}
+
+	streamChannel := make(chan ChatMessageStreamChannelResponse)
+	go api.chatMessagesStreamHandle(ctx, httpResp, streamChannel)
+	return streamChannel, nil
+}
+
+func (api *API) chatMessagesStreamHandle(ctx context.Context, resp *http.Response, streamChannel chan ChatMessageStreamChannelResponse) {
+	defer func(Body io.ReadCloser) {
+		err := Body.Close()
+		if err != nil {
+			fmt.Println(err.Error())
+		}
+	}(resp.Body)
+	defer close(streamChannel)
+
+	reader := bufio.NewReader(resp.Body)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			line, err := reader.ReadBytes('\n')
+			fmt.Println(string(line))
+			if err != nil {
+				streamChannel <- ChatMessageStreamChannelResponse{
+					Err: fmt.Errorf("error reading line: %w", err),
+				}
+				return
+			}
+
+			if !bytes.HasPrefix(line, []byte("data:")) {
+				continue
+			}
+			line = bytes.TrimPrefix(line, []byte("data:"))
+
+			var resp ChatMessageStreamChannelResponse
+			if err = json.Unmarshal(line, &resp); err != nil {
+				streamChannel <- ChatMessageStreamChannelResponse{
+					Err: fmt.Errorf("error unmarshalling event: %w", err),
+				}
+				return
+			} else if resp.Event == "error" {
+				streamChannel <- ChatMessageStreamChannelResponse{
+					Err: errors.New("error streaming event: " + string(line)),
+				}
+				return
+			} else if resp.Answer == "" {
+				return
+			}
+			streamChannel <- resp
+		}
+	}
+}

chat/common/dify/api_chat_stream_test.go

@@ -0,0 +1,50 @@
+package dify
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestChatMessagesStream(t *testing.T) {
+	ctx := context.Background()
+	api := NewClient("https://api.dify.ai", "app-xxxxxxx")
+
+	request := &ChatMessageRequest{
+		Query:  "hi",
+		User:   "xyz",
+		Inputs: map[string]any{},
+	}
+	request.Inputs["session"] = "123"
+	ctx, cancel := context.WithTimeout(ctx, 200*time.Second)
+	defer cancel()
+	var ch chan ChatMessageStreamChannelResponse
+	var err error
+
+	if ch, err = api.API().ChatMessagesStream(ctx, request); err != nil {
+		return
+	}
+	// 设置超时时间为 200 秒
+
+	var strBuilder strings.Builder
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case streamData, isOpen := <-ch:
+			if err = streamData.Err; err != nil {
+				fmt.Println(err.Error())
+				return
+			}
+			if !isOpen {
+				fmt.Println(strBuilder.String())
+				return
+			}
+
+			strBuilder.WriteString(streamData.Answer)
+		}
+	}
+}

chat/common/dify/api_conversations.go

@@ -0,0 +1,77 @@
+package dify
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+)
+
+type ConversationsRequest struct {
+	LastID string `json:"last_id,omitempty"`
+	Limit  int    `json:"limit"`
+	User   string `json:"user"`
+}
+
+type ConversationsResponse struct {
+	Limit   int                         `json:"limit"`
+	HasMore bool                        `json:"has_more"`
+	Data    []ConversationsDataResponse `json:"data"`
+}
+
+type ConversationsDataResponse struct {
+	ID        string            `json:"id"`
+	Name      string            `json:"name"`
+	Inputs    map[string]string `json:"inputs"`
+	Status    string            `json:"status"`
+	CreatedAt int64             `json:"created_at"`
+}
+
+type ConversationsRenamingRequest struct {
+	ConversationID string `json:"conversation_id,omitempty"`
+	Name           string `json:"name"`
+	User           string `json:"user"`
+}
+
+type ConversationsRenamingResponse struct {
+	Result string `json:"result"`
+}
+
+// Conversations /* Get conversation list
+func (api *API) Conversations(ctx context.Context, req *ConversationsRequest) (resp *ConversationsResponse, err error) {
+	if req.User == "" {
+		err = errors.New("ConversationsRequest.User Illegal")
+		return
+	}
+	if req.Limit == 0 {
+		req.Limit = 20
+	}
+
+	httpReq, err := api.createBaseRequest(ctx, http.MethodGet, "/v1/conversations", nil)
+	if err != nil {
+		return
+	}
+
+	query := httpReq.URL.Query()
+	query.Set("last_id", req.LastID)
+	query.Set("user", req.User)
+	query.Set("limit", strconv.FormatInt(int64(req.Limit), 10))
+	httpReq.URL.RawQuery = query.Encode()
+
+	err = api.c.sendJSONRequest(httpReq, &resp)
+	return
+}
+
+// ConversationsRenaming /* Conversation renaming
+func (api *API) ConversationsRenaming(ctx context.Context, req *ConversationsRenamingRequest) (resp *ConversationsRenamingResponse, err error) {
+	url := fmt.Sprintf("/v1/conversations/%s/name", req.ConversationID)
+	req.ConversationID = ""
+
+	httpReq, err := api.createBaseRequest(ctx, http.MethodPost, url, req)
+	if err != nil {
+		return
+	}
+	err = api.c.sendJSONRequest(httpReq, &resp)
+	return
+}