Learn.VideoAnalysis/VideoAnalysisCore/AICore/GPT/DeepSeek/DeepSeekClient.cs

using VideoAnalysisCore.Common;
using System.Net.Http.Headers;
using System.Text;
using Microsoft.Extensions.Logging;
using Newtonsoft.Json.Linq;
using System.Net.Http;
using Newtonsoft.Json;
using System.Net.Http.Json;
using System.Net;
using VideoAnalysisCore.AICore.GPT.KIMI;
using System.Threading;
using System;
using System.IO;
using VideoAnalysisCore.AICore.GPT.ChatGPT;
using System.Threading.Tasks;

namespace VideoAnalysisCore.AICore.GPT.DeepSeek
{

    public class DeepSeekGPTClient
    {

        //private readonly string Path = "v1/chat/completions";
        //public static string Host = AppCommon.Config.ChatGpt.aliyun.Host;
        //public static string ApiKey = AppCommon.Config.ChatGpt.aliyun.ApiKey;

        private readonly string Path = "";
        public static string Host = AppCommon.Config.ChatGpt.DeepSeek.Host;
        public static string ApiKey = AppCommon.Config.ChatGpt.DeepSeek.ApiKey;


        //public static string Host = AppCommon.Config.ChatGpt.ChatGpt.Host;
        //public static string ApiKey = AppCommon.Config.ChatGpt.ChatGpt.ApiKey;
        //private readonly string Path = "v1/chat/completions";

        private readonly IHttpClientFactory _httpClientFactory;

        public DeepSeekGPTClient(IHttpClientFactory httpClientFactory)
        {
            _httpClientFactory = httpClientFactory;
        }


        /// <summary>
        /// Chat
        /// </summary>
        /// <param name="chatReq"></param>
        /// <returns>Return HttpResponseMessage for SSE</returns>
        public async Task<(Usage u, string res,string reasoning)?> Chat(ChatRequest chatReq)
        {
            //chatReq.model = "deepseek-r1";
            if (chatReq.stream)  return await ChatSSE(chatReq);
            postStar:
            var requestBody = System.Text.Json.JsonSerializer.Serialize(chatReq);
            HttpResponseMessage chatResp = PostJsonStream(Path, requestBody);
            var res1 = await chatResp.Content.ReadAsStringAsync();
            if (res1 == null || string.IsNullOrEmpty(res1)|| !chatResp.IsSuccessStatusCode)
            {
                Console.WriteLine(DateTime.Now + $"=>GPT请求失败重试 Code = {chatResp.StatusCode} Res={res1}");
                goto postStar;
            
            }
            //throw new Exception($" GPT模型返回异常 返回参数: " +
            //    $" {System.Text.Json.JsonSerializer.Serialize(res1)}");
            Console.WriteLine(DateTime.Now + $"=>GPT请求头获取成功 Code = {chatResp.StatusCode} Res={res1}");
            var res = await chatResp.Content.ReadFromJsonAsync<ChatRes>();
            if (res is null || res.error != null)
                throw new Exception($" GPT模型返回异常 返回参数: " +
                    $" {System.Text.Json.JsonSerializer.Serialize(res)}");
            var d = thinkMSG(res?.choices.FirstOrDefault()?.message);
            var chatResContent = d.m1;
            var chatResReasoning = d.m2;
           
            if (string.IsNullOrEmpty(chatResContent))
                return null;
            return (res.usage, chatResContent, chatResReasoning);
        }

        private (string m1, string m2) thinkMSG(Message? m) 
        {
            var chatResContent = m?.content.Trim();
            var chatResReasoning = string.Empty;
            if (chatResContent.StartsWith("<think"))
            {
                chatResReasoning = chatResContent.Substring(7, chatResContent.IndexOf("</think") - 7);
                chatResContent = chatResContent
                    .Substring(chatResContent.IndexOf("</think>") + 8);
            }
            else
                chatResReasoning = m?.reasoning_content?.Trim();
            return (chatResContent, chatResReasoning);

        }
        private HttpResponseMessage PostJsonStream(string path, string json)
        {
            var uriBuilder = new UriBuilder(Host + Path);
            var maxRestart = 4;
            var errorMSG = new Exception[maxRestart];
            for (int i = 0; i < maxRestart; i++)
            {
                try
                {
                    var client = _httpClientFactory.CreateClient();
                    client.DefaultRequestHeaders.Authorization =
                        new AuthenticationHeaderValue("Bearer", ApiKey);
                    client.Timeout = TimeSpan.FromSeconds(60 * 20);//超时时间20分钟
                    client.DefaultRequestVersion = HttpVersion.Version20;
                    client.DefaultVersionPolicy = HttpVersionPolicy.RequestVersionOrLower;
                    client.DefaultRequestHeaders.ConnectionClose = true;

                    var request = new HttpRequestMessage(HttpMethod.Post, uriBuilder.Uri);
                    request.Content = new StringContent(json, Encoding.UTF8, "application/json");

                    return client.Send(request, HttpCompletionOption.ResponseHeadersRead);
                }
                catch (Exception e)
                {
                    errorMSG[i] = e;
                    Console.WriteLine("====================[请求异常,重试]====================");
                    Console.WriteLine(uriBuilder.Uri);
                    Console.WriteLine(e.Message);
                    Console.WriteLine(e.StackTrace);
                    Console.WriteLine("==============================================");

                Thread.Sleep(1000);
                }
            }
            throw errorMSG.Last(s => s != null);
        }




        /// <summary>
        /// ChatSSE[流式传输 更稳定]
        /// </summary>
        /// <param name="chatReq"></param>
        /// <returns>Return HttpResponseMessage for SSE</returns>
        public async Task<(Usage u, string res, string reasoning)?> ChatSSE(ChatRequest chatReq)
        {
            var requestBody = System.Text.Json.JsonSerializer.Serialize(chatReq);
            PostJsonStream:
            var chatResp = PostJsonStream(string.Empty, requestBody);
            if (!chatResp.IsSuccessStatusCode) 
            {
                Console.WriteLine(DateTime.Now + "=>请求GPT服务器异常 " + chatResp?.StatusCode);
                Console.WriteLine(await chatResp.Content.ReadAsStringAsync());
                goto PostJsonStream;
            }
            using var stream = chatResp.Content.ReadAsStream();
            using var reader = new StreamReader(stream, Encoding.UTF8);
            string line;
            var messageBuilder = new StringBuilder();
            var messageBuilder1 = new StringBuilder();
            var lastChat = new ChatResSSE();
            var splitCount = "data:".Length;
            var maxLoop = 60*1000;
            int threshold = 0;
            while (maxLoop>0)
            {
                line = reader.ReadLine();
                if (line is null || string.IsNullOrEmpty(line)|| line.StartsWith(": keep-alive")) {
                    Thread.Sleep(100);
                    maxLoop--;
                    continue; 
                }
                else if (line.EndsWith("[DONE]"))
                {
                    // 表示一条消息结束
                    string message = messageBuilder.ToString();
                    string message2 = messageBuilder1.ToString();
                    messageBuilder.Clear();
                    messageBuilder1.Clear();
                    var d =thinkMSG(new Message() { content = message, reasoning_content = message2 });
                    message = d.m1;
                    message2 = d.m2;
                    var u = lastChat?.usage;
                    if (u == null || string.IsNullOrEmpty(message))
                        return null;
                    return (u, message, message2);
                }
                else if (line.StartsWith("data:"))
                {
                    try
                    {
                        var data = System.Text.Json.JsonSerializer.Deserialize<ChatResSSE>(line.Substring(splitCount).Trim());
                        lastChat = data;
                        var delta = data?.choices.FirstOrDefault()?.delta;
                        var str = delta?.content;
                        var strReasoning = delta?.reasoning_content;
                        if (!string.IsNullOrEmpty(str))
                            messageBuilder.Append(str);
                        if (!string.IsNullOrEmpty(strReasoning))
                            messageBuilder1.Append(strReasoning);
                        var steamCount = messageBuilder.Length + messageBuilder1.Length;
                        if (++threshold%30==0)
                            RedisExpand.SetTaskProgress(chatReq.taskId, "steam=>"+ steamCount);
                    }
                    catch (Exception e)
                    {
                        Console.WriteLine("异常 ChatSSE=>");
                        Console.WriteLine(line);
                        Console.WriteLine(e.Message);
                        Console.WriteLine(e.StackTrace);
                    }
                }
            }
            return null;
        }

    }
}