Root / src / Dpz.Core.Service.Network / ChatCompletionService.cs

using Dpz.Core.Entity.Base;
using Dpz.Core.Service.Network.Models;

namespace Dpz.Core.Service.Network;

/// <summary>
/// Chat Completion API 服务，用于调用通用的AI ChatCompletion接口
/// 支持 ChatGPT、Claude、Gemini、DeepSeek 等多个AI提供商
/// </summary>
[Obsolete("use IOpenAiService")]
public class ChatCompletionService(
    IConfiguration configuration,
    ILogger<ChatCompletionService> logger,
    HttpClient httpClient
)
{
    private readonly Lazy<string> _host = new(() =>
        configuration["ChatCompletionHost"]
        ?? throw new InvalidConfigurationException("ChatCompletionHost configuration is missing")
    );

    private readonly Lazy<string> _apiKey = new(() =>
        configuration["ChatCompletionApiKey"]
        ?? throw new InvalidConfigurationException("ChatCompletionApiKey configuration is missing")
    );

    /// <summary>
    /// 发送聊天消息
    /// 流式或非流式响应取决于ChatCompletionOption中的Stream属性
    /// 如果是流式响应，返回原始流式数据；如果是非流式，返回解析后的ChatCompletionResponse
    /// </summary>
    /// <param name="messages">消息列表</param>
    /// <param name="options">API请求选项配置</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>API响应结果，根据是否流式返回不同类型</returns>
    public async Task<ResponseResult<ChatCompletionResponse?>> SendChatAsync(
        List<ChatMessage> messages,
        Action<ChatCompletionOption>? options = null,
        CancellationToken cancellationToken = default
    )
    {
        if (messages is not { Count: > 0 })
        {
            logger.LogWarning("Message list is empty");
            return new ResponseResult<ChatCompletionResponse?>().WithFail(
                "Messages cannot be empty"
            );
        }

        logger.LogInformation("Starting chat request with {MessageCount} messages", messages.Count);

        var callOption =
            configuration.GetSection("DefaultChatCompletionOption").Get<ChatCompletionOption>()
            ?? new ChatCompletionOption();

        options?.Invoke(callOption);

        // 根据配置决定是否使用流式响应
        if (callOption.Stream)
        {
            return await SendStreamChatAsync(messages, callOption, cancellationToken);
        }

        return await SendRequestAsync(messages, callOption, cancellationToken);
    }

    /// <summary>
    /// 发送流式聊天请求并返回解析后的响应（一次性读取完整内容后再拆分）
    /// </summary>
    private async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatAsync(
        List<ChatMessage> messages,
        ChatCompletionOption option,
        CancellationToken cancellationToken = default
    )
    {
        var result = new ResponseResult<ChatCompletionResponse?>();

        try
        {
            var rawResponse = await SendRawRequestAsync(messages, option, cancellationToken);
            if (!rawResponse.Success)
            {
                logger.LogWarning("Stream chat request failed: {Message}", rawResponse.Message);
                return result.WithFail(rawResponse.Message ?? "Request failed");
            }

            var content = rawResponse.Data;
            if (string.IsNullOrEmpty(content))
            {
                logger.LogWarning("Stream response content is empty");
                return result.WithFail("Response content is empty");
            }

            // 解析流式响应
            var parsedResponse = ParseStreamResponse(content);

            if (parsedResponse == null)
            {
                logger.LogWarning("Failed to parse stream response");
                return result.WithFail("Failed to parse stream response");
            }

            logger.LogInformation(
                "Stream chat completed - ID: {ResponseId}, Model: {Model}",
                parsedResponse.Id,
                parsedResponse.Model
            );
            return result.WithOk(parsedResponse);
        }
        catch (Exception e)
        {
            logger.LogError(e, "Stream chat request exception");
            return result.WithFail($"Exception: {e.Message}");
        }
    }

    /// <summary>
    /// 解析流式响应，将所有块合并为单个响应
    /// </summary>
    private ChatCompletionResponse? ParseStreamResponse(string rawContent)
    {
        try
        {
            var lines = rawContent.Split(["\n", "\r\n"], StringSplitOptions.RemoveEmptyEntries);
            var parser = new StreamResponseParser(logger);
            return parser.Parse(lines);
        }
        catch (Exception e)
        {
            logger.LogError(e, "Parse stream response exception");
            return null;
        }
    }

    /// <summary>
    /// 发送请求并返回解析后的响应
    /// </summary>
    private async Task<ResponseResult<ChatCompletionResponse?>> SendRequestAsync(
        List<ChatMessage> messages,
        ChatCompletionOption option,
        CancellationToken cancellationToken = default
    )
    {
        var result = new ResponseResult<ChatCompletionResponse?>();

        try
        {
            var rawResponse = await SendRawRequestAsync(messages, option, cancellationToken);
            if (!rawResponse.Success)
            {
                logger.LogWarning("Non-stream chat request failed: {Message}", rawResponse.Message);
                return result.WithFail(rawResponse.Message ?? "Request failed");
            }

            var content = rawResponse.Data;
            if (string.IsNullOrEmpty(content))
            {
                logger.LogWarning("Non-stream response content is empty");
                return result.WithFail("Response content is empty");
            }

            var jsonOptions = new JsonSerializerOptions { PropertyNameCaseInsensitive = true };
            var response = JsonSerializer.Deserialize<ChatCompletionResponse>(content, jsonOptions);

            if (response == null)
            {
                var truncatedContent = content[..Math.Min(200, content.Length)];
                logger.LogWarning(
                    "Failed to deserialize response. Content (first 200 chars): {Content}",
                    truncatedContent
                );
                return result.WithFail("Failed to deserialize response");
            }

            logger.LogInformation(
                "Non-stream chat completed - ID: {ResponseId}, Model: {Model}, Tokens: {PromptTokens}/{CompletionTokens}/{TotalTokens}",
                response.Id,
                response.Model,
                response.Usage?.PromptTokens ?? 0,
                response.Usage?.CompletionTokens ?? 0,
                response.Usage?.TotalTokens ?? 0
            );
            return result.WithOk(response);
        }
        catch (Exception e)
        {
            logger.LogError(e, "Non-stream chat request exception");
            return result.WithFail($"Exception: {e.Message}");
        }
    }

    /// <summary>
    /// 发送原始请求并返回未解析的响应内容（缓冲模式）
    /// </summary>
    private async Task<ResponseResult<string?>> SendRawRequestAsync(
        List<ChatMessage> messages,
        ChatCompletionOption option,
        CancellationToken cancellationToken = default
    )
    {
        var result = new ResponseResult<string?>();
        var request = new HttpRequestMessage(HttpMethod.Post, $"{_host.Value}/v2/chat/completions");

        request.Headers.Add("Authorization", $"Basic {_apiKey.Value}");

        var requestBody = BuildRequestBody(messages, option);
        var jsonOptions = new JsonSerializerOptions
        {
            DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
        };

        logger.LogDebug(
            "Sending API request - URL: {Url}, Model: {Model}, Stream: {Stream}",
            request.RequestUri,
            ((Dictionary<string, object?>)requestBody)["model"],
            ((Dictionary<string, object?>)requestBody)["stream"]
        );

        var jsonContent = JsonSerializer.Serialize(requestBody, jsonOptions);
        request.Content = new StringContent(jsonContent, Encoding.UTF8, "application/json");

        HttpResponseMessage response;
        try
        {
            response = await httpClient.SendAsync(request, cancellationToken);
            if (!response.IsSuccessStatusCode)
            {
                var errorContent = await response.Content.ReadAsStringAsync(cancellationToken);
                var truncatedError = errorContent[..Math.Min(500, errorContent.Length)];
                logger.LogWarning(
                    "API call failed - Status: {StatusCode}, Error: {ErrorContent}",
                    response.StatusCode,
                    truncatedError
                );
                return result.WithFail(
                    $"API call failed with status {response.StatusCode}: {errorContent}"
                );
            }
        }
        catch (OperationCanceledException)
        {
            logger.LogInformation("HTTP request was cancelled");
            throw;
        }
        catch (Exception e)
        {
            logger.LogError(e, "HTTP request exception");
            return result.WithFail($"HTTP request exception: {e.Message}");
        }

        var content = await response.Content.ReadAsStringAsync(cancellationToken);
        logger.LogDebug(
            "API response received - Content length: {ContentLength} chars",
            content.Length
        );

        return result.WithOk(content);
    }

    /// <summary>
    /// 构建请求体，只包含有值的字段
    /// </summary>
    private static object BuildRequestBody(List<ChatMessage> messages, ChatCompletionOption option)
    {
        // 当启用thinking功能时，temperature必须为1.0
        var temperature = option.Temperature;
        if (option.Thinking != null)
        {
            temperature = 1.0;
        }

        var body = new Dictionary<string, object?>
        {
            { "messages", messages },
            { "model", AiModelMapper.GetModelString(option.Model) },
            { "stream", option.Stream },
            { "max_tokens", option.MaxTokens },
            { "temperature", temperature },
        };

        if (option.FrequencyPenalty.HasValue)
        {
            body["frequency_penalty"] = option.FrequencyPenalty;
        }

        if (option.TopP.HasValue)
        {
            body["top_p"] = option.TopP;
        }

        if (option.Thinking != null)
        {
            body["thinking"] = new
            {
                type = option.Thinking.Type,
                budget_tokens = option.Thinking.BudgetTokens,
            };
        }

        return body;
    }

    /// <summary>
    /// 发送流式聊天请求，通过回调逐步推送内容增量
    /// </summary>
    /// <param name="messages">消息列表</param>
    /// <param name="onDeltaContent">当收到内容增量时的回调</param>
    /// <param name="options">API请求选项配置</param>
    /// <param name="cancellationToken">取消令牌</param>
    /// <returns>完整的API响应结果</returns>
    public async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatWithCallbackAsync(
        List<ChatMessage> messages,
        Func<string, Task> onDeltaContent,
        Action<ChatCompletionOption>? options = null,
        CancellationToken cancellationToken = default
    )
    {
        if (messages is not { Count: > 0 })
        {
            logger.LogWarning("Message list is empty");
            return new ResponseResult<ChatCompletionResponse?>().WithFail(
                "Messages cannot be empty"
            );
        }

        logger.LogInformation("Starting stream chat request with callback");

        var callOption =
            configuration.GetSection("DefaultChatCompletionOption").Get<ChatCompletionOption>()
            ?? new ChatCompletionOption();

        options?.Invoke(callOption);

        // 强制启用流式模式
        callOption.Stream = true;

        return await SendStreamChatHttpIncrementalAsync(
            messages,
            callOption,
            onDeltaContent,
            cancellationToken
        );
    }

    /// <summary>
    /// HTTP增量读取实现：使用 ResponseHeadersRead 和逐行解析SSE
    /// </summary>
    private async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatHttpIncrementalAsync(
        List<ChatMessage> messages,
        ChatCompletionOption option,
        Func<string, Task> onDeltaContent,
        CancellationToken cancellationToken = default
    )
    {
        var result = new ResponseResult<ChatCompletionResponse?>();

        // 在开始前检查取消
        cancellationToken.ThrowIfCancellationRequested();

        var request = new HttpRequestMessage(HttpMethod.Post, $"{_host.Value}/v2/chat/completions");
        request.Headers.Add("Authorization", $"Basic {_apiKey.Value}");

        var requestBody = BuildRequestBody(messages, option);
        var jsonOptions = new JsonSerializerOptions
        {
            DefaultIgnoreCondition = JsonIgnoreCondition.WhenWritingNull,
        };
        var jsonContent = JsonSerializer.Serialize(requestBody, jsonOptions);
        request.Content = new StringContent(jsonContent, Encoding.UTF8, "application/json");

        HttpResponseMessage response;
        try
        {
            response = await httpClient.SendAsync(
                request,
                HttpCompletionOption.ResponseHeadersRead,
                cancellationToken
            );
            if (!response.IsSuccessStatusCode)
            {
                var errorContent = await response.Content.ReadAsStringAsync(cancellationToken);
                var truncatedError = errorContent[..Math.Min(300, errorContent.Length)];
                logger.LogWarning(
                    "Streaming API call failed - Status: {StatusCode}, Error: {Error}",
                    response.StatusCode,
                    truncatedError
                );
                return result.WithFail(
                    $"API call failed with status {response.StatusCode}: {truncatedError}"
                );
            }
        }
        catch (OperationCanceledException)
        {
            logger.LogInformation("Streaming HTTP request was cancelled");
            throw;
        }
        catch (Exception ex)
        {
            logger.LogError(ex, "Streaming HTTP request exception");
            return result.WithFail($"HTTP request exception: {ex.Message}");
        }

        await using var stream = await response.Content.ReadAsStreamAsync(cancellationToken);
        using var reader = new StreamReader(stream, Encoding.UTF8);

        var accumulator = new StreamAccumulator();
        // 累积同一事件的多行 data:
        var sbEvent = new StringBuilder();
        var jsonDeserializeOptions = new JsonSerializerOptions
        {
            PropertyNameCaseInsensitive = true,
        };

        async Task ProcessEventAsync()
        {
            // 在处理事件前检查取消
            cancellationToken.ThrowIfCancellationRequested();

            if (sbEvent.Length == 0)
            {
                return;
            }

            var jsonStr = sbEvent.ToString().Trim();
            sbEvent.Clear();
            if (string.IsNullOrEmpty(jsonStr))
            {
                return;
            }

            if (jsonStr == "[DONE]")
            {
                // 正常结束标记
                return;
            }

            StreamCompletionChunk? chunk = null;
            try
            {
                chunk = JsonSerializer.Deserialize<StreamCompletionChunk>(
                    jsonStr,
                    jsonDeserializeOptions
                );
            }
            catch (JsonException je)
            {
                var truncated = jsonStr[..Math.Min(120, jsonStr.Length)];
                logger.LogWarning(je, "Failed to deserialize streaming chunk: {Chunk}", truncated);
            }

            if (chunk == null)
            {
                return;
            }

            if (!accumulator.IsInitialized)
            {
                accumulator.Initialize(chunk);
            }

            var delta = accumulator.AccumulateChunk(chunk);
            if (!string.IsNullOrEmpty(delta))
            {
                // 在回调前再次检查取消
                cancellationToken.ThrowIfCancellationRequested();

                try
                {
                    await onDeltaContent(delta);
                }
                catch (OperationCanceledException)
                {
                    // 回调中的取消操作需要向上传播
                    throw;
                }
                catch (Exception callbackEx)
                {
                    logger.LogWarning(callbackEx, "Delta callback threw an exception");
                }
            }
        }

        try
        {
            string? line;
            while ((line = await reader.ReadLineAsync(cancellationToken)) != null)
            {
                // 在每次循环开始时立即检查取消令牌
                cancellationToken.ThrowIfCancellationRequested();

                if (string.IsNullOrEmpty(line))
                {
                    // 事件结束分隔（空行）
                    await ProcessEventAsync();
                    continue;
                }

                // SSE注释行或非data行直接忽略
                if (line.StartsWith(":"))
                {
                    continue;
                }

                if (!line.StartsWith("data:"))
                {
                    // 忽略非data行
                    continue;
                }

                var payload = line["data:".Length..].TrimStart();

                // 结束标记可立即处理（清空缓冲并跳出）
                if (payload == "[DONE]")
                {
                    // 先处理已有累积块
                    await ProcessEventAsync();
                    break;
                }

                // 累积当前事件的 data 行（可能多行）
                sbEvent.Append(payload);
            }

            // 处理可能剩余的事件
            await ProcessEventAsync();
        }
        catch (OperationCanceledException)
        {
            logger.LogInformation("Stream processing was cancelled");
            throw;
        }

        var final = accumulator.BuildFinalResponse();
        if (final == null)
        {
            return result.WithFail("Failed to build final streaming response");
        }

        logger.LogInformation(
            "Streaming chat completed - ID: {Id}, Model: {Model}",
            final.Id,
            final.Model
        );
        return result.WithOk(final);
    }
}

⚠⚠ 以下内容为AI分析的结果，请根据实际情况进行判断。

代码解释

这是一个 AI 聊天完成服务 (ChatCompletionService) 的实现，用于调用各种 AI 提供商（如 ChatGPT、Claude、Gemini、DeepSeek）的 Chat Completion API。该服务已被标记为 [Obsolete]，建议使用 IOpenAiService 替代。

核心功能

1. 服务初始化

public class ChatCompletionService(
    IConfiguration configuration,
    ILogger<ChatCompletionService> logger,
    HttpClient httpClient
)

使用主构造函数注入依赖
通过 Lazy<T> 延迟加载配置项（ChatCompletionHost 和 ChatCompletionApiKey）
如果配置缺失会抛出 InvalidConfigurationException

2. 主要接口方法

SendChatAsync - 通用聊天接口

public async Task<ResponseResult<ChatCompletionResponse?>> SendChatAsync(
    List<ChatMessage> messages,
    Action<ChatCompletionOption>? options = null,
    CancellationToken cancellationToken = default
)

功能：

验证消息列表非空
加载默认配置并允许通过 Action<ChatCompletionOption> 自定义
根据 callOption.Stream 决定调用流式或非流式请求

SendStreamChatWithCallbackAsync - 流式聊天（带回调）

public async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatWithCallbackAsync(
    List<ChatMessage> messages,
    Func<string, Task> onDeltaContent,
    Action<ChatCompletionOption>? options = null,
    CancellationToken cancellationToken = default
)

功能：

强制启用流式模式
通过 onDeltaContent 回调实时推送每个内容增量（适用于逐字显示场景）
最终返回完整的聊天响应

3. 内部实现方法

SendStreamChatAsync - 流式聊天（一次性读取）

private async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatAsync(...)

调用 SendRawRequestAsync 获取原始响应
使用 ParseStreamResponse 解析流式数据（按行拆分后解析）

SendRequestAsync - 非流式聊天

private async Task<ResponseResult<ChatCompletionResponse?>> SendRequestAsync(...)

调用 SendRawRequestAsync 获取 JSON 响应
直接反序列化为 ChatCompletionResponse 对象
记录 Token 使用量（PromptTokens/CompletionTokens/TotalTokens）

SendRawRequestAsync - 发送 HTTP 请求（缓冲模式）

private async Task<ResponseResult<string?>> SendRawRequestAsync(...)

关键步骤：

构造 HTTP POST 请求到 /v2/chat/completions
添加 Authorization: Basic {apiKey} 头
调用 BuildRequestBody 构造请求体
使用 JsonSerializer 序列化并发送
返回完整响应内容（缓冲读取）

SendStreamChatHttpIncrementalAsync - 增量流式读取

private async Task<ResponseResult<ChatCompletionResponse?>> SendStreamChatHttpIncrementalAsync(...)

核心实现（SSE 流式解析）：

使用 HttpCompletionOption.ResponseHeadersRead 立即返回头部（不等待完整响应）
使用 StreamReader 逐行读取 SSE（Server-Sent Events）数据
处理 data: 行：
- 忽略注释行（: 开头）
- 累积同一事件的多行 data:
- 遇到空行或 [DONE] 时处理完整事件
使用 StreamAccumulator 累积增量数据
通过 onDeltaContent 回调实时推送每个内容增量
支持取消令牌（在关键位置检查 cancellationToken.ThrowIfCancellationRequested()）

BuildRequestBody - 构造请求体

private static object BuildRequestBody(List<ChatMessage> messages, ChatCompletionOption option)

特殊处理：

当启用 thinking 功能时，强制 temperature = 1.0
只包含有值的可选字段（frequency_penalty、top_p）
通过 AiModelMapper.GetModelString 映射模型名称

ParseStreamResponse - 解析流式响应

private ChatCompletionResponse? ParseStreamResponse(string rawContent)

按行拆分原始内容
使用 StreamResponseParser 解析成完整的 ChatCompletionResponse

关键设计模式

1. 懒加载模式

private readonly Lazy<string> _host = new(() => 
    configuration["ChatCompletionHost"] ?? throw new InvalidConfigurationException(...)
);

延迟到首次访问时才读取配置，提高启动性能

2. 流式构建器模式

StreamAccumulator 累积增量数据
支持实时推送和最终合并

3. 统一错误处理

所有方法返回 ResponseResult<T>（包含 Success、Message、Data）
日志记录覆盖警告、错误和调试信息

4. 取消支持

在关键循环和异步操作中检查 CancellationToken
区分 OperationCanceledException（正常取消）和其他异常

配置依赖

需要在 appsettings.json 中配置：

{
  "ChatCompletionHost": "https://api.example.com",
  "ChatCompletionApiKey": "your-api-key",
  "DefaultChatCompletionOption": {
    "Model": "gpt-4",
    "Stream": false,
    "MaxTokens": 4000,
    "Temperature": 0.7
  }
}

总结

这是一个功能完善的 AI 聊天服务封装，支持：

✅ 流式/非流式响应
✅ 实时增量回调
✅ 多 AI 提供商兼容
✅ 完善的错误处理和日志
✅ 取消令牌支持
✅ SSE（Server-Sent Events）流式解析

主要应用场景：AI 聊天机器人、内容生成、智能问答系统等。

评论加载中...