xwl2023/Yi.Framework
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: 完成openai响应接口

Browse Source
This commit is contained in:
chenchun
2025-12-11 17:16:21 +08:00
parent 433d616b9b
commit 69b84f6613
6 changed files with 379 additions and 137 deletions
Download Patch File Download Diff File Expand all files Collapse all files

195
Yi.Abp.Net8/module/ai-hub/Yi.Framework.AiHub.Domain/Managers/AiGateWayManager.cs
View File

@@ -21,6 +21,7 @@ using Yi.Framework.AiHub.Domain.Shared.Dtos.OpenAi.Embeddings;
using Yi.Framework.AiHub.Domain.Shared.Dtos.OpenAi.Images;
using Yi.Framework.AiHub.Domain.Shared.Dtos.OpenAi.Responses;
using Yi.Framework.AiHub.Domain.Shared.Enums;
using Yi.Framework.AiHub.Domain.Shared.Extensions;
using Yi.Framework.Core.Extensions;
using Yi.Framework.SqlSugarCore.Abstractions;
using JsonSerializer = System.Text.Json.JsonSerializer;
@@ -91,30 +92,7 @@ public class AiGateWayManager : DomainService
return aiModelDescribe;
}
/// <summary>
/// 聊天完成-流式
/// </summary>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async IAsyncEnumerable<ThorChatCompletionsResponse> CompleteChatStreamAsync(
ThorChatCompletionsRequest request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
_specialCompatible.Compatible(request);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.OpenAi, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IChatCompletionService>(modelDescribe.HandlerName);
await foreach (var result in chatService.CompleteChatStreamAsync(modelDescribe, request, cancellationToken))
{
result.SupplementalMultiplier(modelDescribe.Multiplier);
yield return result;
}
}
/// <summary>
/// 聊天完成-非流式
/// </summary>
@@ -176,6 +154,7 @@ public class AiGateWayManager : DomainService
await response.WriteAsJsonAsync(data, cancellationToken);
}
/// <summary>
/// 聊天完成-缓存处理
/// </summary>
@@ -201,8 +180,12 @@ public class AiGateWayManager : DomainService
response.Headers.TryAdd("Connection", "keep-alive");
var gateWay = LazyServiceProvider.GetRequiredService<AiGateWayManager>();
var completeChatResponse = gateWay.CompleteChatStreamAsync(request, cancellationToken);
_specialCompatible.Compatible(request);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.OpenAi, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IChatCompletionService>(modelDescribe.HandlerName);
var completeChatResponse = chatService.CompleteChatStreamAsync(modelDescribe,request, cancellationToken);
var tokenUsage = new ThorUsageResponse();
//缓存队列算法
@@ -244,6 +227,7 @@ public class AiGateWayManager : DomainService
{
await foreach (var data in completeChatResponse)
{
data.SupplementalMultiplier(modelDescribe.Multiplier);
if (data.Usage is not null && (data.Usage.CompletionTokens > 0 || data.Usage.OutputTokens > 0))
{
tokenUsage = data.Usage;
@@ -316,8 +300,8 @@ public class AiGateWayManager : DomainService
}
}
}
/// <summary>
/// 图片生成
/// </summary>
@@ -385,8 +369,8 @@ public class AiGateWayManager : DomainService
throw new UserFriendlyException(errorContent);
}
}
/// <summary>
/// 向量生成
/// </summary>
@@ -498,30 +482,7 @@ public class AiGateWayManager : DomainService
throw new UserFriendlyException(errorContent);
}
}
/// <summary>
/// Anthropic聊天完成-流式
/// </summary>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async IAsyncEnumerable<(string, AnthropicStreamDto?)> AnthropicCompleteChatStreamAsync(
AnthropicInput request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
_specialCompatible.AnthropicCompatible(request);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.Claude, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IAnthropicChatCompletionService>(modelDescribe.HandlerName);
await foreach (var result in chatService.StreamChatCompletionsAsync(modelDescribe, request, cancellationToken))
{
result.Item2.SupplementalMultiplier(modelDescribe.Multiplier);
yield return result;
}
}
/// <summary>
/// Anthropic聊天完成-非流式
@@ -582,6 +543,7 @@ public class AiGateWayManager : DomainService
await response.WriteAsJsonAsync(data, cancellationToken);
}
/// <summary>
/// Anthropic聊天完成-缓存处理
/// </summary>
@@ -605,16 +567,20 @@ public class AiGateWayManager : DomainService
response.ContentType = "text/event-stream;charset=utf-8;";
response.Headers.TryAdd("Cache-Control", "no-cache");
response.Headers.TryAdd("Connection", "keep-alive");
var gateWay = LazyServiceProvider.GetRequiredService<AiGateWayManager>();
var completeChatResponse = gateWay.AnthropicCompleteChatStreamAsync(request, cancellationToken);
_specialCompatible.AnthropicCompatible(request);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.Claude, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IAnthropicChatCompletionService>(modelDescribe.HandlerName);
var completeChatResponse = chatService.StreamChatCompletionsAsync(modelDescribe,request, cancellationToken);
ThorUsageResponse? tokenUsage = null;
StringBuilder backupSystemContent = new StringBuilder();
try
{
await foreach (var responseResult in completeChatResponse)
{
responseResult.Item2.SupplementalMultiplier(modelDescribe.Multiplier);
//message_start是为了保底机制
if (responseResult.Item1.Contains("message_delta") || responseResult.Item1.Contains("message_start"))
{
@@ -679,7 +645,6 @@ public class AiGateWayManager : DomainService
Guid? tokenId = null,
CancellationToken cancellationToken = default)
{
//todo 1
// _specialCompatible.AnthropicCompatible(request);
var response = httpContext.Response;
// 设置响应头，声明是 json
@@ -689,69 +654,47 @@ public class AiGateWayManager : DomainService
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IOpenAiResponseService>(modelDescribe.HandlerName);
var data = await chatService.ResponsesAsync(modelDescribe, request, cancellationToken);
//todo 2
// data.SupplementalMultiplier(modelDescribe.Multiplier);
//todo 3
// if (userId is not null)
// {
// await _aiMessageManager.CreateUserMessageAsync(userId.Value, sessionId,
// new MessageInputDto
// {
// Content = sessionId is null ? "不予存储" : request.Messages?.FirstOrDefault()?.Content ?? string.Empty,
// ModelId = request.Model,
// TokenUsage = data.TokenUsage,
// }, tokenId);
//
// await _aiMessageManager.CreateSystemMessageAsync(userId.Value, sessionId,
// new MessageInputDto
// {
// Content = sessionId is null ? "不予存储" : data.content?.FirstOrDefault()?.text,
// ModelId = request.Model,
// TokenUsage = data.TokenUsage
// }, tokenId);
//
// await _usageStatisticsManager.SetUsageAsync(userId.Value, request.Model, data.TokenUsage, tokenId);
//
// // 扣减尊享token包用量
// var totalTokens = data.TokenUsage.TotalTokens ?? 0;
// if (totalTokens > 0)
// {
// await PremiumPackageManager.TryConsumeTokensAsync(userId.Value, totalTokens);
// }
// }
data.SupplementalMultiplier(modelDescribe.Multiplier);
var tokenUsage= new ThorUsageResponse
{
InputTokens = data.Usage.InputTokens,
OutputTokens = data.Usage.OutputTokens,
TotalTokens = data.Usage.InputTokens + data.Usage.OutputTokens,
};
if (userId is not null)
{
await _aiMessageManager.CreateUserMessageAsync(userId.Value, sessionId,
new MessageInputDto
{
Content = "不予存储",
ModelId = request.Model,
TokenUsage = tokenUsage,
}, tokenId);
await _aiMessageManager.CreateSystemMessageAsync(userId.Value, sessionId,
new MessageInputDto
{
Content = "不予存储",
ModelId = request.Model,
TokenUsage = tokenUsage
}, tokenId);
await _usageStatisticsManager.SetUsageAsync(userId.Value, request.Model, tokenUsage, tokenId);
// 扣减尊享token包用量
var totalTokens = tokenUsage.TotalTokens ?? 0;
if (totalTokens > 0)
{
await PremiumPackageManager.TryConsumeTokensAsync(userId.Value, totalTokens);
}
}
await response.WriteAsJsonAsync(data, cancellationToken);
}
/// <summary>
/// OpenAi 响应-流式
/// </summary>
/// <param name="request"></param>
/// <param name="cancellationToken"></param>
/// <returns></returns>
public async IAsyncEnumerable<(string, dynamic?)> OpenAiResponsesAsync(
OpenAiResponsesInput request,
[EnumeratorCancellation] CancellationToken cancellationToken)
{
//todo cc
// _specialCompatible.AnthropicCompatible(request);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.Response, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IOpenAiResponseService>(modelDescribe.HandlerName);
await foreach (var result in chatService.ResponsesStreamAsync(modelDescribe, request, cancellationToken))
{
//todo 倍率
// result.Item2.SupplementalMultiplier(modelDescribe.Multiplier);
yield return result;
}
}
/// <summary>
/// OpenAi响应-流式-缓存处理
/// </summary>
@@ -776,7 +719,11 @@ public class AiGateWayManager : DomainService
response.Headers.TryAdd("Cache-Control", "no-cache");
response.Headers.TryAdd("Connection", "keep-alive");
var completeChatResponse = OpenAiResponsesAsync(request, cancellationToken);
var modelDescribe = await GetModelAsync(ModelApiTypeEnum.Response, request.Model);
var chatService =
LazyServiceProvider.GetRequiredKeyedService<IOpenAiResponseService>(modelDescribe.HandlerName);
var completeChatResponse = chatService.ResponsesStreamAsync(modelDescribe,request, cancellationToken);
ThorUsageResponse? tokenUsage = null;
try
{
@@ -785,12 +732,14 @@ public class AiGateWayManager : DomainService
//message_start是为了保底机制
if (responseResult.Item1.Contains("response.completed"))
{
JObject obj = JObject.FromObject(responseResult.Item2);
int inputTokens = (int?)obj["response"]?["usage"]?["input_tokens"] ?? 0;
int outputTokens = (int?)obj["response"]?["usage"]?["output_tokens"] ?? 0;
var obj = responseResult.Item2!.Value;
int inputTokens = obj.GetPath("response","usage","input_tokens").GetInt();
int outputTokens = obj.GetPath("response","usage","output_tokens").GetInt();
inputTokens=Convert.ToInt32(inputTokens * modelDescribe.Multiplier);
outputTokens=Convert.ToInt32(outputTokens * modelDescribe.Multiplier);
tokenUsage = new ThorUsageResponse
{
PromptTokens = inputTokens,
PromptTokens =inputTokens,
InputTokens = inputTokens,
OutputTokens = outputTokens,
CompletionTokens = outputTokens,
@@ -839,7 +788,7 @@ public class AiGateWayManager : DomainService
}
#region Anthropic格式Http响应
#region 流式传输格式Http响应
private static readonly byte[] EventPrefix = "event: "u8.ToArray();
private static readonly byte[] DataPrefix = "data: "u8.ToArray();