refactor(rag): 重构AI对话和工作流集成实现

- 移除IWorkflowTriggerService接口及其实现,改用Mediator模式直接发送命令
- 更新AiBotHandler移除workflowTrigger依赖,使用新命令模式触发工作流
- 修改ApproveTaskEndpoint和RejectTaskEndpoint返回bool类型并使用Send.OkAsync
- 更新appsettings.json中AI助手的默认指令配置
- 在BuildFlowHostModule.cs中移除旧的工作流触发服务注册
- 添加SSE流式响应路径前缀配置以支持RAG聊天流
- 重构ChatAgentService使用HTTP客户端工厂和运行时配置解析
- 实现AiChatOutputSanitizer用于清洗AI输出中的模板内容
- 更新EmbeddingService使用运行时配置解析器动态创建客户端
- 调整RAG查询处理器中的系统提示指令内容
- 添加构建流按钮操作的单元测试验证发布和禁用功能
- 扩展即时消息集成测试覆盖文件消息元数据处理场景
- 更新平台权限种子数据添加按钮发布和禁用权限
- 实现AI模型运行时配置解析器支持动态模型切换
This commit is contained in:
xiangning 2026-07-04 21:29:37 +08:00
parent d6dc6be51d
commit 54dc3fc26e
46 changed files with 1485 additions and 233 deletions

View File

@ -21,7 +21,6 @@ using Volo.Abp;
using Volo.Abp.Modularity;
using Workflow.Application;
using Workflow.Application.Engine;
using Workflow.Application.Integration;
using Workflow.Application.Notifications;
using Workflow.Infrastructure;
@ -90,8 +89,6 @@ public class BuildFlowHostModule : AbpModule
// 串联:工作流通知 → 平台消息中心铃铛(跨模块集成,Host 实现 IPlatformNotificationPublisher 供 workflow 调用,
// 内部转发为 PublishNotificationCommand 复用 platform 的 fanout 落库 + SignalR 推送)
services.AddScoped<IPlatformNotificationPublisher, WorkflowNotificationPublisher>();
// 串联:AI 对话 → 触发工作流(跨模块集成,Workflow 实现 IWorkflowTriggerService 供 RAG 调用)
services.AddScoped<BuildFlow.Shared.Integration.IWorkflowTriggerService, WorkflowTriggerService>();
// workflow 后台服务:超时调度 + Webhook 投递
services.AddScoped<ProcessEngine>();
services.Configure<NotificationOptions>(config.GetSection("Workflow:Notifications"));
@ -122,6 +119,9 @@ public class BuildFlowHostModule : AbpModule
opts.PassthroughPrefixes.Add("/hubs");
// Swagger/OpenAPI 文档必须保持原始结构,不能包裹成 { code, data, message },否则 Swagger UI 无法识别 openapi 字段。
opts.PassthroughPrefixes.Add("/swagger");
// RAG/Chat SSE 必须直连响应流。若进入响应包裹中间件的 MemoryStream,FastEndpoints 会在响应未启动时收尾成 204。
opts.StreamPathPrefixes.Add("/api/chat/conversations");
opts.StreamPathPrefixes.Add("/api/rag");
});
}

View File

@ -0,0 +1,24 @@
using BuildFlow.Application.Document.DTOs;
using BuildFlow.Application.Document.Queries;
using MediatR;
namespace BuildFlow.Host.Endpoints.Rag.Document;
/// <summary>
/// 预览文档切片端点。只返回按当前策略切出的片段,不触发向量化和持久化。
/// </summary>
public class PreviewDocumentChunksEndpoint(IMediator mediator) : EndpointWithoutRequest<List<ChunkPreviewDto>>
{
public override void Configure()
{
Get("/documents/{DocumentId}/chunks/preview");
AllowAnonymous();
}
public override async Task HandleAsync(CancellationToken ct)
{
var docId = Route<Guid>("DocumentId");
var result = await mediator.Send(new PreviewDocumentChunksQuery(docId), ct);
await Send.OkAsync(result, ct);
}
}

View File

@ -68,7 +68,7 @@ public class RAGStreamEndpoint(
var ragMessages = new List<AiChatMessage>
{
new(AiChatRole.System, "基于提供的上下文回答问题。如果上下文中没有相关信息,请如实说明。"),
new(AiChatRole.System, "基于提供的上下文直接回答问题。不要输出身份开场、Markdown标题或推理过程;上下文没有依据时,只用一句话说明未检索到相关内容。"),
new(AiChatRole.User, $"{contextBuilder}\n\n问题:{req.Question}"),
};

View File

@ -0,0 +1,22 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.Commands;
using Workflow.Application.BuildFlow.Buttons.DTOs;
namespace BuildFlow.Host.Endpoints.Workflow.BuildFlow.Buttons;
/// <summary>禁用 BuildFlow 按钮定义端点。POST /buildflow/buttons/{id}/disable。</summary>
public class DisableBuildFlowButtonDefinitionEndpoint(IMediator mediator)
: Endpoint<DisableBuildFlowButtonDefinitionRequest, BuildFlowButtonDefinitionDto>
{
public override void Configure()
{
Post("/buildflow/buttons/{id}/disable");
Permissions("buildflow:button:disable");
}
public override async Task HandleAsync(DisableBuildFlowButtonDefinitionRequest req, CancellationToken ct)
{
var result = await mediator.Send(new DisableBuildFlowButtonDefinitionCommand(req.Id), ct);
await Send.OkAsync(result, ct);
}
}

View File

@ -0,0 +1,4 @@
namespace BuildFlow.Host.Endpoints.Workflow.BuildFlow.Buttons;
/// <summary>禁用 BuildFlow 按钮定义请求。</summary>
public sealed record DisableBuildFlowButtonDefinitionRequest(Guid Id);

View File

@ -0,0 +1,22 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.Commands;
using Workflow.Application.BuildFlow.Buttons.DTOs;
namespace BuildFlow.Host.Endpoints.Workflow.BuildFlow.Buttons;
/// <summary>发布 BuildFlow 按钮定义端点。POST /buildflow/buttons/{id}/publish。</summary>
public class PublishBuildFlowButtonDefinitionEndpoint(IMediator mediator)
: Endpoint<PublishBuildFlowButtonDefinitionRequest, BuildFlowButtonDefinitionDto>
{
public override void Configure()
{
Post("/buildflow/buttons/{id}/publish");
Permissions("buildflow:button:publish");
}
public override async Task HandleAsync(PublishBuildFlowButtonDefinitionRequest req, CancellationToken ct)
{
var result = await mediator.Send(new PublishBuildFlowButtonDefinitionCommand(req.Id), ct);
await Send.OkAsync(result, ct);
}
}

View File

@ -0,0 +1,4 @@
namespace BuildFlow.Host.Endpoints.Workflow.BuildFlow.Buttons;
/// <summary>发布 BuildFlow 按钮定义请求。</summary>
public sealed record PublishBuildFlowButtonDefinitionRequest(Guid Id);

View File

@ -7,7 +7,7 @@ namespace BuildFlow.Host.Endpoints.Workflow.WorkflowTask;
/// <summary>
/// 审批通过任务端点。驱动 TaskStateMachine 转 Approved,按 Approved 边推进流程。
/// </summary>
public class ApproveTaskEndpoint : Endpoint<ApproveTaskRequest>
public class ApproveTaskEndpoint : Endpoint<ApproveTaskRequest, bool>
{
private readonly IMediator _mediator;
@ -27,8 +27,6 @@ public class ApproveTaskEndpoint : Endpoint<ApproveTaskRequest>
{
var command = new ApproveTaskCommand(req.Id, req.UserId, req.Comment, req.FormDataJson);
await _mediator.Send(command, ct);
HttpContext.Response.StatusCode = 200;
HttpContext.Response.ContentType = "application/json";
await HttpContext.Response.WriteAsync("{}", ct);
await Send.OkAsync(true, ct);
}
}

View File

@ -7,7 +7,7 @@ namespace BuildFlow.Host.Endpoints.Workflow.WorkflowTask;
/// <summary>
/// 驳回任务端点。转 Rejected 终态,按 Rejected 边流转(若无则结束实例)。
/// </summary>
public class RejectTaskEndpoint : Endpoint<RejectTaskRequest>
public class RejectTaskEndpoint : Endpoint<RejectTaskRequest, bool>
{
private readonly IMediator _mediator;
@ -27,8 +27,6 @@ public class RejectTaskEndpoint : Endpoint<RejectTaskRequest>
{
var command = new RejectTaskCommand(req.Id, req.UserId, req.Comment);
await _mediator.Send(command, ct);
HttpContext.Response.StatusCode = 200;
HttpContext.Response.ContentType = "application/json";
await HttpContext.Response.WriteAsync("{}", ct);
await Send.OkAsync(true, ct);
}
}

View File

@ -65,7 +65,7 @@
"ApiKey": "lm-studio",
"ChatModel": "qwen/qwen3.6-27b",
"EmbeddingModel": "text-embedding-embeddinggemma-300m",
"DefaultInstructions": "你是一个RAG知识库助手,基于提供的上下文准确回答用户问题。如果上下文中没有相关信息,请如实说明。",
"DefaultInstructions": "你是 BuildFlow AI 助手。普通对话直接回答;当消息明确包含知识库上下文时,优先基于上下文回答。不要输出“作为RAG知识库助手”等身份开场、Markdown标题或推理过程;不要因为普通对话没有上下文就要求用户补充资料。",
"MaxTokens": 2000,
"Temperature": 0.7
}

View File

@ -144,6 +144,8 @@ public static class SeedData
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000051"), "读取按钮", "buildflow:button:read", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000052"), "创建按钮", "buildflow:button:create", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000053"), "执行按钮", "buildflow:button:execute", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000059"), "发布按钮", "buildflow:button:publish", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000060"), "禁用按钮", "buildflow:button:disable", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000054"), "读取函数", "buildflow:function:read", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000055"), "创建函数", "buildflow:function:create", "BuildFlow"),
new SeedPermission(Guid.Parse("10000000-0000-0000-0000-000000000056"), "发布函数", "buildflow:function:publish", "BuildFlow"),

View File

@ -18,5 +18,6 @@
<ItemGroup>
<ProjectReference Include="..\Domain\BuildFlow.Domain.csproj" />
<ProjectReference Include="..\Infrastructure\BuildFlow.Infrastructure.csproj" />
<ProjectReference Include="..\..\workflow\Application\Workflow.Application.csproj" />
</ItemGroup>
</Project>

View File

@ -0,0 +1,89 @@
using BuildFlow.Application.Document.DTOs;
using BuildFlow.Application.Document.Queries;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Shared.Domain.Exceptions;
using MediatR;
using Microsoft.AspNetCore.Http;
namespace BuildFlow.Application.Document.Handlers;
/// <summary>
/// 预览文档切片查询处理器。复用正式处理的提取器和分块策略,但不调用 embedding,也不持久化 chunk。
/// </summary>
public class PreviewDocumentChunksQueryHandler(
RagDbContext db,
IDocumentExtractorFactory extractorFactory,
ITextChunker chunker,
IFileStorageClient fileStorage,
IHttpContextAccessor httpContextAccessor)
: IRequestHandler<PreviewDocumentChunksQuery, List<ChunkPreviewDto>>
{
public async Task<List<ChunkPreviewDto>> Handle(PreviewDocumentChunksQuery request, CancellationToken ct)
{
var doc = await db.Documents.FindAsync([request.DocumentId], ct)
?? throw new NotFoundException("文档不存在");
var kb = await db.KnowledgeBases.FindAsync([doc.KnowledgeBaseId], ct)
?? throw new NotFoundException("知识库不存在");
string? tempDownloadPath = null;
try
{
var extractPath = await ResolveFilePathAsync(doc, ct);
tempDownloadPath = extractPath == doc.FilePath ? null : extractPath;
var extractor = extractorFactory.GetExtractor(doc.FileName);
var extracted = await extractor.ExtractAsync(extractPath, ct);
if (string.IsNullOrWhiteSpace(extracted.Content))
throw new BusinessException($"文件「{doc.FileName}」提取到的内容为空,无法预览切片");
var effectiveMode = doc.ChunkingModeOverride ?? kb.ChunkingMode;
var chunks = chunker.Chunk(extracted, effectiveMode, kb.ChunkSize, kb.ChunkOverlap, kb.Separator);
return chunks.Select(chunk => new ChunkPreviewDto(
Guid.NewGuid(),
chunk.Index,
chunk.Content,
chunk.Content.Length / 4,
false,
null,
null,
chunk.HeadingPath,
null)).ToList();
}
finally
{
if (tempDownloadPath is not null && File.Exists(tempDownloadPath))
{
try { File.Delete(tempDownloadPath); } catch { /* 忽略清理失败 */ }
}
}
}
/// <summary>
/// 解析文档文件路径。本地路径直接返回;文件服务对象下载到临时文件供提取器读取。
/// </summary>
private async Task<string> ResolveFilePathAsync(BuildFlow.Domain.Entities.Document doc, CancellationToken ct)
{
if (File.Exists(doc.FilePath))
return doc.FilePath;
var slashIdx = doc.FilePath.IndexOf('/');
if (slashIdx <= 0 || slashIdx >= doc.FilePath.Length - 1)
throw new BusinessException($"文档路径格式无效:{doc.FilePath}");
var bucket = doc.FilePath[..slashIdx];
var objectKey = doc.FilePath[(slashIdx + 1)..];
var authToken = httpContextAccessor.HttpContext?.Request.Headers.Authorization.ToString().Replace("Bearer ", "");
var ext = Path.GetExtension(doc.FileName);
var tempPath = Path.Combine(Path.GetTempPath(), $"rag_preview_{Guid.NewGuid()}{ext}");
await using (var stream = await fileStorage.DownloadAsync(bucket, objectKey, authToken, ct))
await using (var fs = File.Create(tempPath))
{
await stream.CopyToAsync(fs, ct);
}
return tempPath;
}
}

View File

@ -0,0 +1,9 @@
using MediatR;
using BuildFlow.Application.Document.DTOs;
namespace BuildFlow.Application.Document.Queries;
/// <summary>
/// 预览文档切片查询。只执行提取和分块,不生成向量、不写入正式分块表。
/// </summary>
public record PreviewDocumentChunksQuery(Guid DocumentId) : IRequest<List<ChunkPreviewDto>>;

View File

@ -2,6 +2,8 @@ using MediatR;
using BuildFlow.Application.RagQA.Commands;
using BuildFlow.Infrastructure.Persistence;
using BuildFlow.Shared.Integration;
using Workflow.Application.Integration.Commands;
using Workflow.Application.Integration.Queries;
namespace BuildFlow.Application.Integration;
@ -12,19 +14,15 @@ namespace BuildFlow.Application.Integration;
/// </summary>
public class AiBotHandler(
IMediator mediator,
RagDbContext db,
IWorkflowTriggerService? workflowTrigger = null) : IAiBotHandler
RagDbContext db) : IAiBotHandler
{
public async Task<string> AnswerAsync(string question, Guid conversationId, CancellationToken ct)
{
// 串联链路4:AI 对话 → 触发工作流
// 检测"发起/启动/申请 XXX 流程"等意图
if (workflowTrigger is not null)
{
var triggerResult = await TryDetectWorkflowIntentAsync(question, ct);
if (triggerResult is not null)
return triggerResult;
}
var triggerResult = await TryDetectWorkflowIntentAsync(question, ct);
if (triggerResult is not null)
return triggerResult;
// 默认:RAG 知识库问答
var kb = await db.KnowledgeBases
@ -71,7 +69,7 @@ public class AiBotHandler(
if (string.IsNullOrWhiteSpace(flowName)) return null;
var result = await workflowTrigger!.TryStartAsync(flowName, Guid.Empty, null, ct);
var result = await mediator.Send(new StartWorkflowByNameCommand(flowName, Guid.Empty, null), ct);
if (result.Success)
{
@ -80,7 +78,7 @@ public class AiBotHandler(
else
{
// 列出可用流程供用户选择
var defs = await workflowTrigger.GetTriggerableDefinitionsAsync(ct);
var defs = await mediator.Send(new GetTriggerableWorkflowDefinitionsQuery(), ct);
if (defs.Count > 0)
{
var list = string.Join("\n", defs.Select(d => $" • {d.Name}"));

View File

@ -45,11 +45,11 @@ public class RagQueryCommandHandler(
// 用完整命名空间避免与 BuildFlow.Domain.Entities.ChatMessage 歧义
var ragMessages = new List<Microsoft.Extensions.AI.ChatMessage>
{
new(ChatRole.System, "基于提供的上下文回答问题。如果上下文中没有相关信息,请如实说明。"),
new(ChatRole.System, "基于提供的上下文直接回答问题。不要输出身份开场、Markdown标题或推理过程;上下文没有依据时,只用一句话说明未检索到相关内容。"),
new(ChatRole.User, $"{contextBuilder}\n\n问题:{request.Question}"),
};
var answer = await chatAgent.RunAsync(ragMessages, ct);
return new RAGQueryResponse(answer, sources);
}
}
}

View File

@ -0,0 +1,209 @@
using System.Text;
namespace BuildFlow.Infrastructure.AI;
/// <summary>
/// AI 对话输出清洗器。过滤思考标签和模型常见身份声明,避免模板话术进入用户可见气泡。
/// </summary>
public sealed class AiChatOutputSanitizer
{
private const string FallbackQuestionPrompt = "请先输入要查询的问题,或选择知识库后再提问。";
private static readonly string[] RagAssistantOpenings =
[
"您好!作为RAG知识库助手",
"您好,作为RAG知识库助手",
"作为RAG知识库助手",
"您好!我是RAG知识库助手",
"您好,我是RAG知识库助手",
"我是RAG知识库助手",
];
private readonly StringBuilder _buffer = new();
private readonly StringBuilder _openingBuffer = new();
private bool _inThinkTag;
private bool _openingResolved;
private bool _suppressAsBoilerplate;
/// <summary>
/// 追加一段模型输出,返回可立即推送给前端的清洗后片段。
/// </summary>
public IEnumerable<string> Append(string text)
{
if (string.IsNullOrEmpty(text)) yield break;
_buffer.Append(text);
foreach (var visible in DrainVisibleText(false))
{
foreach (var chunk in DrainOpening(visible))
yield return chunk;
}
}
/// <summary>
/// 结束流式读取,返回尚未输出的安全尾部内容。
/// </summary>
public IEnumerable<string> Complete()
{
foreach (var visible in DrainVisibleText(true))
{
foreach (var chunk in DrainOpening(visible))
yield return chunk;
}
if (_suppressAsBoilerplate)
{
yield return FallbackQuestionPrompt;
yield break;
}
if (!_openingResolved && _openingBuffer.Length > 0)
yield return _openingBuffer.ToString();
_openingBuffer.Clear();
}
/// <summary>
/// 清洗完整回复文本,供非流式路径或持久化前兜底使用。
/// </summary>
public static string SanitizeFinalText(string text)
{
if (string.IsNullOrWhiteSpace(text)) return string.Empty;
var withoutThink = RemoveThinkTags(text).TrimStart('\n', '\r', ' ');
return IsBoilerplateQuestionPrompt(withoutThink)
? FallbackQuestionPrompt
: withoutThink;
}
private IEnumerable<string> DrainVisibleText(bool flushAll)
{
while (_buffer.Length > 0)
{
var content = _buffer.ToString();
if (_inThinkTag)
{
var endIdx = content.IndexOf("</think", StringComparison.OrdinalIgnoreCase);
if (endIdx < 0)
{
if (flushAll) _buffer.Clear();
else if (_buffer.Length > 10) _buffer.Remove(0, _buffer.Length - 10);
yield break;
}
var closeIdx = content.IndexOf('>', endIdx);
if (closeIdx < 0) yield break;
_inThinkTag = false;
_buffer.Remove(0, closeIdx + 1);
continue;
}
var thinkIdx = content.IndexOf("<think", StringComparison.OrdinalIgnoreCase);
if (thinkIdx < 0)
{
if (!flushAll && _buffer.Length <= 7) yield break;
var safeLen = flushAll ? _buffer.Length : _buffer.Length - 7;
var output = _buffer.ToString(0, safeLen);
_buffer.Remove(0, safeLen);
if (!string.IsNullOrEmpty(output)) yield return output;
continue;
}
if (thinkIdx > 0)
yield return _buffer.ToString(0, thinkIdx);
var endIdx2 = content.IndexOf("</think", thinkIdx, StringComparison.OrdinalIgnoreCase);
if (endIdx2 >= 0)
{
var closeIdx = content.IndexOf('>', endIdx2);
if (closeIdx < 0) yield break;
_buffer.Remove(0, closeIdx + 1);
}
else
{
_inThinkTag = true;
_buffer.Remove(0, thinkIdx);
yield break;
}
}
}
private IEnumerable<string> DrainOpening(string text)
{
if (_suppressAsBoilerplate) yield break;
if (_openingResolved)
{
yield return text;
yield break;
}
_openingBuffer.Append(text);
var opening = _openingBuffer.ToString();
if (IsBoilerplateQuestionPrompt(opening))
{
_suppressAsBoilerplate = true;
_openingBuffer.Clear();
yield break;
}
if (MightBecomeBoilerplate(opening)) yield break;
_openingResolved = true;
yield return opening;
_openingBuffer.Clear();
}
private static string RemoveThinkTags(string text)
{
var span = text.AsSpan();
var builder = new StringBuilder();
while (!span.IsEmpty)
{
var thinkStart = span.IndexOf("<think".AsSpan(), StringComparison.OrdinalIgnoreCase);
if (thinkStart < 0)
{
builder.Append(span);
break;
}
builder.Append(span[..thinkStart]);
var afterThink = span[(thinkStart + 6)..];
var gtIdx = afterThink.IndexOf('>');
if (gtIdx < 0) break;
var body = afterThink[(gtIdx + 1)..];
var endTag = body.IndexOf("</think".AsSpan(), StringComparison.OrdinalIgnoreCase);
if (endTag < 0) break;
var afterEnd = body[endTag..];
var endGt = afterEnd.IndexOf('>');
if (endGt < 0) break;
span = afterEnd[(endGt + 1)..];
}
return builder.ToString();
}
private static bool IsBoilerplateQuestionPrompt(string text)
{
var normalized = text.Replace("**", "", StringComparison.Ordinal);
return normalized.Contains("RAG知识库助手", StringComparison.OrdinalIgnoreCase)
&& (normalized.Contains("具体的问题", StringComparison.Ordinal)
|| normalized.Contains("知识库内容", StringComparison.Ordinal)
|| normalized.Contains("请问您想了解什么", StringComparison.Ordinal)
|| normalized.Contains("请随时补充信息", StringComparison.Ordinal));
}
private static bool MightBecomeBoilerplate(string text)
{
if (text.Length < 20 && text.StartsWith("您好", StringComparison.Ordinal))
return true;
return text.Length < 120 && RagAssistantOpenings.Any(opening =>
opening.StartsWith(text, StringComparison.Ordinal)
|| text.StartsWith(opening, StringComparison.Ordinal));
}
}

View File

@ -0,0 +1,12 @@
namespace BuildFlow.Infrastructure.AI;
/// <summary>
/// AI 运行时模型配置。封装一次 OpenAI 兼容调用所需的模型 ID、端点和密钥。
/// </summary>
public record AiModelRuntimeConfig(
string ModelId,
string BaseUrl,
string ApiKey,
string? DefaultInstructions,
int MaxTokens,
double Temperature);

View File

@ -0,0 +1,67 @@
using BuildFlow.Domain.Enums;
using BuildFlow.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
namespace BuildFlow.Infrastructure.AI;
/// <summary>
/// AI 运行时模型配置解析器。优先读取模型管理中的默认启用模型,没有配置时回退到 appsettings。
/// </summary>
public class AiModelRuntimeConfigResolver(
IServiceScopeFactory scopeFactory,
IOptions<AiOptions> options)
{
private readonly AiOptions _fallbackOptions = options.Value;
/// <summary>
/// 按模型类型解析本次调用配置。Chat 与 Embedding 都通过该入口,避免某一类调用绕过本地模型配置。
/// </summary>
public async Task<AiModelRuntimeConfig> ResolveAsync(ModelType modelType, CancellationToken ct)
{
using var scope = scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<RagDbContext>();
var model = await db.AiModels
.AsNoTracking()
.FirstOrDefaultAsync(m => m.ModelType == modelType && m.IsDefault && m.IsActive, ct);
if (model is not null)
{
return new AiModelRuntimeConfig(
model.ModelId,
model.BaseUrl,
model.ApiKey,
ResolveDefaultInstructions(model.DefaultInstructions),
model.MaxTokens,
model.Temperature);
}
var fallbackModelId = modelType == ModelType.Embedding
? _fallbackOptions.EmbeddingModel
: _fallbackOptions.ChatModel;
return new AiModelRuntimeConfig(
fallbackModelId,
_fallbackOptions.BaseUrl,
_fallbackOptions.ApiKey,
_fallbackOptions.DefaultInstructions,
_fallbackOptions.MaxTokens,
_fallbackOptions.Temperature);
}
private string? ResolveDefaultInstructions(string? modelInstructions)
{
if (string.IsNullOrWhiteSpace(modelInstructions))
return _fallbackOptions.DefaultInstructions;
return IsLegacyRagOnlyInstructions(modelInstructions)
? _fallbackOptions.DefaultInstructions
: modelInstructions;
}
private static bool IsLegacyRagOnlyInstructions(string instructions) =>
instructions.Contains("RAG知识库助手", StringComparison.OrdinalIgnoreCase)
&& instructions.Contains("上下文", StringComparison.Ordinal)
&& instructions.Contains("没有相关信息", StringComparison.Ordinal);
}

View File

@ -1,137 +1,96 @@
using System.Runtime.CompilerServices;
using System.Text;
using System.ClientModel;
using Microsoft.Agents.AI;
using Microsoft.EntityFrameworkCore;
using System.Net.Http.Headers;
using Microsoft.Extensions.AI;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using OpenAI;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.Persistence;
using ModelType = BuildFlow.Domain.Enums.ModelType;
namespace BuildFlow.Infrastructure.AI;
public class ChatAgentService : IAIChatAgent
/// <summary>
/// AI 对话代理服务。按模型管理中的默认 Chat 模型调用 OpenAI 兼容 Chat Completions 接口。
/// </summary>
public class ChatAgentService(
AiModelRuntimeConfigResolver configResolver,
IHttpClientFactory httpClientFactory)
: IAIChatAgent
{
private readonly IServiceScopeFactory _scopeFactory;
private readonly AiOptions _fallbackOptions;
public ChatAgentService(IServiceScopeFactory scopeFactory, IOptions<AiOptions> options)
{
_scopeFactory = scopeFactory;
_fallbackOptions = options.Value;
}
private async Task<(string ModelId, string BaseUrl, string ApiKey, string Instructions)> ResolveConfigAsync(CancellationToken ct)
{
using var scope = _scopeFactory.CreateScope();
var db = scope.ServiceProvider.GetRequiredService<RagDbContext>();
var model = await db.AiModels
.AsNoTracking()
.FirstOrDefaultAsync(m => m.ModelType == ModelType.Chat && m.IsDefault && m.IsActive && !m.IsDeleted, ct);
if (model is not null)
return (model.ModelId, model.BaseUrl, model.ApiKey, model.DefaultInstructions ?? _fallbackOptions.DefaultInstructions);
return (_fallbackOptions.ChatModel, _fallbackOptions.BaseUrl, _fallbackOptions.ApiKey, _fallbackOptions.DefaultInstructions);
}
private async Task<AIAgent> CreateAgentAsync(CancellationToken ct)
{
var (modelId, baseUrl, apiKey, instructions) = await ResolveConfigAsync(ct);
var client = new OpenAIClient(
new ApiKeyCredential(apiKey),
new OpenAIClientOptions { Endpoint = new Uri(baseUrl) });
IChatClient chatClient = client.GetChatClient(modelId).AsIChatClient();
return chatClient.AsAIAgent(name: "BuildFlowAgent", instructions: instructions);
}
private const string ResponseStyleInstructions = """
回答规则:
1. 直接回答用户问题,不要输出“您好!作为RAG知识库助手”“下面是我的回答”等开场白。
2. 不要输出 Markdown 一级/二级标题,除非用户明确要求标题或提纲。
3. 不要展示推理过程、思考标签或 reasoning 内容。
4. 普通对话没有知识库上下文时,按通用 AI 助手直接回答,不要说上下文不足。
5. 如果用户没有提出明确问题,只用一句话提示其输入问题,不要自我介绍。
""";
public async Task<string> RunAsync(IReadOnlyList<ChatMessage> messages, CancellationToken ct = default)
{
var agent = await CreateAgentAsync(ct);
var response = await agent.RunAsync(messages, cancellationToken: ct);
return FilterThinkTags(response.ToString());
var reply = new StringBuilder();
await foreach (var chunk in RunStreamingAsync(messages, ct))
reply.Append(chunk);
return AiChatOutputSanitizer.SanitizeFinalText(reply.ToString());
}
public async IAsyncEnumerable<string> RunStreamingAsync(
IReadOnlyList<ChatMessage> messages,
[EnumeratorCancellation] CancellationToken ct = default)
{
var agent = await CreateAgentAsync(ct);
var buffer = new StringBuilder();
var inThink = false;
var config = await configResolver.ResolveAsync(ModelType.Chat, ct);
using var request = BuildRequest(config, messages);
var client = httpClientFactory.CreateClient();
await foreach (var update in agent.RunStreamingAsync(messages, cancellationToken: ct))
using var response = await client.SendAsync(
request,
HttpCompletionOption.ResponseHeadersRead,
ct);
response.EnsureSuccessStatusCode();
await using var responseStream = await response.Content.ReadAsStreamAsync(ct);
using var reader = new StreamReader(responseStream);
var sanitizer = new AiChatOutputSanitizer();
while (await reader.ReadLineAsync(ct) is { } line)
{
var text = update.ToString();
if (string.IsNullOrEmpty(text)) continue;
buffer.Append(text);
if (!line.StartsWith("data:", StringComparison.OrdinalIgnoreCase))
continue;
while (buffer.Length > 0)
{
var content = buffer.ToString();
if (inThink)
{
var endIdx = content.IndexOf("</think", StringComparison.Ordinal);
if (endIdx < 0) { if (buffer.Length > 10) buffer.Remove(0, buffer.Length - 10); break; }
var closeIdx = content.IndexOf('>', endIdx);
if (closeIdx < 0) break;
inThink = false;
buffer.Remove(0, closeIdx + 1);
}
else
{
var thinkIdx = content.IndexOf("<think", StringComparison.Ordinal);
if (thinkIdx < 0)
{
if (buffer.Length > 7)
{
var safeLen = buffer.Length - 7;
var output = buffer.ToString(0, safeLen);
buffer.Remove(0, safeLen);
if (output.Length > 0) yield return output;
}
break;
}
if (thinkIdx > 0) yield return buffer.ToString(0, thinkIdx);
var endIdx = content.IndexOf("</think", thinkIdx, StringComparison.Ordinal);
if (endIdx >= 0)
{
var closeIdx = content.IndexOf('>', endIdx);
if (closeIdx >= 0) { buffer.Remove(0, closeIdx + 1); }
else { inThink = true; buffer.Clear(); break; }
}
else { inThink = true; buffer.Remove(0, thinkIdx); break; }
}
}
var data = line[5..].Trim();
var content = OpenAiCompatibleChatStreamParser.ReadContentDelta(data);
if (content is null)
continue;
foreach (var chunk in sanitizer.Append(content))
yield return chunk;
}
if (!inThink && buffer.Length > 0) yield return buffer.ToString();
buffer.Clear();
foreach (var chunk in sanitizer.Complete())
yield return chunk;
}
private static string FilterThinkTags(string text)
private static HttpRequestMessage BuildRequest(
AiModelRuntimeConfig config,
IReadOnlyList<ChatMessage> messages)
{
if (string.IsNullOrEmpty(text)) return text;
var sb = new StringBuilder();
var span = text.AsSpan();
while (!span.IsEmpty)
{
var thinkStart = span.IndexOf("<think".AsSpan());
if (thinkStart < 0) { sb.Append(span); break; }
sb.Append(span[..thinkStart]);
var afterThink = span[(thinkStart + 6)..];
var gtIdx = afterThink.IndexOf('>');
if (gtIdx < 0) break;
var endTag = afterThink[(gtIdx + 1)..].IndexOf("</think".AsSpan());
if (endTag < 0) break;
var afterEnd = afterThink[(gtIdx + 1 + endTag)..];
var endGt = afterEnd.IndexOf('>');
if (endGt < 0) break;
span = afterEnd[(endGt + 1)..];
}
return sb.ToString().TrimStart('\n', '\r');
var request = new HttpRequestMessage(HttpMethod.Post, BuildChatCompletionsUri(config.BaseUrl));
if (!string.IsNullOrWhiteSpace(config.ApiKey))
request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", config.ApiKey);
request.Content = new StringContent(
OpenAiCompatibleChatRequestPayload.BuildJson(config, messages, ResponseStyleInstructions),
Encoding.UTF8,
"application/json");
return request;
}
private static Uri BuildChatCompletionsUri(string baseUrl)
{
var normalized = baseUrl.TrimEnd('/');
if (normalized.EndsWith("/chat/completions", StringComparison.OrdinalIgnoreCase))
return new Uri(normalized);
return new Uri($"{normalized}/chat/completions");
}
}

View File

@ -1,7 +1,7 @@
using System.ClientModel;
using Microsoft.Extensions.Options;
using OpenAI;
using OpenAI.Embeddings;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
namespace BuildFlow.Infrastructure.AI;
@ -19,30 +19,26 @@ public class EmbeddingService : IEmbeddingService
/// </summary>
private const int EmbeddingBatchSize = 64;
private readonly EmbeddingClient _client;
private readonly AiModelRuntimeConfigResolver _configResolver;
public EmbeddingService(IOptions<AiOptions> options)
{
var opt = options.Value;
var openaiClient = new OpenAIClient(
new ApiKeyCredential(opt.ApiKey),
new OpenAIClientOptions { Endpoint = new Uri(opt.BaseUrl) });
_client = openaiClient.GetEmbeddingClient(opt.EmbeddingModel);
}
public EmbeddingService(AiModelRuntimeConfigResolver configResolver)
=> _configResolver = configResolver;
public async Task<float[]> EmbedAsync(string text, CancellationToken ct)
{
var response = await _client.GenerateEmbeddingAsync(text, cancellationToken: ct);
var client = await CreateClientAsync(ct);
var response = await client.GenerateEmbeddingAsync(text, cancellationToken: ct);
return response.Value.ToFloats().ToArray();
}
public async Task<List<float[]>> EmbedBatchAsync(List<string> texts, CancellationToken ct)
{
var client = await CreateClientAsync(ct);
// 文本量小则一次调用即可,避免小请求也走分批的循环开销
if (texts.Count <= EmbeddingBatchSize)
{
return await EmbedOneBatchAsync(texts, ct);
return await EmbedOneBatchAsync(client, texts, ct);
}
// 大批量分批调用并按原顺序拼接结果。上层(ProcessDocumentCommand)按索引对应 chunk,
@ -51,14 +47,24 @@ public class EmbeddingService : IEmbeddingService
for (var i = 0; i < texts.Count; i += EmbeddingBatchSize)
{
var slice = texts.GetRange(i, Math.Min(EmbeddingBatchSize, texts.Count - i));
results.AddRange(await EmbedOneBatchAsync(slice, ct));
results.AddRange(await EmbedOneBatchAsync(client, slice, ct));
}
return results;
}
private async Task<List<float[]>> EmbedOneBatchAsync(List<string> texts, CancellationToken ct)
private async Task<EmbeddingClient> CreateClientAsync(CancellationToken ct)
{
var response = await _client.GenerateEmbeddingsAsync(texts, cancellationToken: ct);
var config = await _configResolver.ResolveAsync(ModelType.Embedding, ct);
var openaiClient = new OpenAIClient(
new ApiKeyCredential(config.ApiKey),
new OpenAIClientOptions { Endpoint = new Uri(config.BaseUrl) });
return openaiClient.GetEmbeddingClient(config.ModelId);
}
private static async Task<List<float[]>> EmbedOneBatchAsync(EmbeddingClient client, List<string> texts, CancellationToken ct)
{
var response = await client.GenerateEmbeddingsAsync(texts, cancellationToken: ct);
return response.Value.Select(e => e.ToFloats().ToArray()).ToList();
}
}

View File

@ -0,0 +1,66 @@
using System.Text.Json;
using Microsoft.Extensions.AI;
namespace BuildFlow.Infrastructure.AI;
/// <summary>
/// OpenAI 兼容 Chat Completions 请求载荷构造器。集中控制 BuildFlow 对本地模型的生成参数。
/// </summary>
public static class OpenAiCompatibleChatRequestPayload
{
/// <summary>
/// 构造 JSON 请求体。reasoning_effort=none 用于避免 Qwen/DeepSeek 类模型长时间只输出推理流。
/// </summary>
public static string BuildJson(
AiModelRuntimeConfig config,
IReadOnlyList<ChatMessage> messages,
string responseStyleInstructions)
{
var payload = new
{
model = config.ModelId,
messages = BuildOpenAiMessages(config, messages, responseStyleInstructions),
stream = true,
temperature = config.Temperature,
max_tokens = config.MaxTokens,
reasoning_effort = "none",
};
return JsonSerializer.Serialize(payload);
}
private static List<object> BuildOpenAiMessages(
AiModelRuntimeConfig config,
IReadOnlyList<ChatMessage> messages,
string responseStyleInstructions)
{
var result = new List<object>
{
new { role = "system", content = BuildSystemInstructions(config.DefaultInstructions, responseStyleInstructions) },
};
result.AddRange(messages.Select(m => new
{
role = ToOpenAiRole(m.Role),
content = m.Text,
}));
return result;
}
private static string BuildSystemInstructions(string? configuredInstructions, string responseStyleInstructions)
{
if (string.IsNullOrWhiteSpace(configuredInstructions))
return responseStyleInstructions;
return $"{configuredInstructions.Trim()}\n\n{responseStyleInstructions}";
}
private static string ToOpenAiRole(ChatRole role)
{
if (role == ChatRole.System) return "system";
if (role == ChatRole.Assistant) return "assistant";
if (role == ChatRole.Tool) return "tool";
return "user";
}
}

View File

@ -0,0 +1,42 @@
using System.Text.Json;
namespace BuildFlow.Infrastructure.AI;
/// <summary>
/// OpenAI 兼容 Chat Completions SSE 解析器。LM Studio 的 /v1/chat/completions 流式响应同样使用该结构。
/// </summary>
public static class OpenAiCompatibleChatStreamParser
{
/// <summary>
/// 从单条 SSE data 内容中读取可展示正文增量。reasoning_content 属于推理过程,不进入最终气泡。
/// </summary>
public static string? ReadContentDelta(string data)
{
if (string.IsNullOrWhiteSpace(data) || data.Trim() == "[DONE]")
return null;
using var document = JsonDocument.Parse(data);
if (!document.RootElement.TryGetProperty("choices", out var choices)
|| choices.ValueKind != JsonValueKind.Array
|| choices.GetArrayLength() == 0)
{
return null;
}
var choice = choices[0];
if (!choice.TryGetProperty("delta", out var delta)
|| delta.ValueKind != JsonValueKind.Object)
{
return null;
}
if (delta.TryGetProperty("content", out var content)
&& content.ValueKind == JsonValueKind.String)
{
var text = content.GetString();
return string.IsNullOrEmpty(text) ? null : text;
}
return null;
}
}

View File

@ -44,6 +44,7 @@ public class RAGInfrastructureModule : AbpModule
// AI 服务注册
services.Configure<AiOptions>(config.GetSection(AiOptions.SectionName));
services.AddScoped<AiModelRuntimeConfigResolver>();
services.AddScoped<IAIChatAgent, ChatAgentService>();
services.AddScoped<IEmbeddingService, EmbeddingService>();
services.AddScoped<ITextChunker, TextChunker>();

View File

@ -0,0 +1,8 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.DTOs;
namespace Workflow.Application.BuildFlow.Buttons.Commands;
/// <summary>禁用 BuildFlow 按钮定义命令。禁用后运行时不再加载该按钮。</summary>
public sealed record DisableBuildFlowButtonDefinitionCommand(Guid Id)
: IRequest<BuildFlowButtonDefinitionDto>;

View File

@ -0,0 +1,8 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.DTOs;
namespace Workflow.Application.BuildFlow.Buttons.Commands;
/// <summary>发布 BuildFlow 按钮定义命令。发布后运行时才能加载并执行按钮。</summary>
public sealed record PublishBuildFlowButtonDefinitionCommand(Guid Id)
: IRequest<BuildFlowButtonDefinitionDto>;

View File

@ -0,0 +1,44 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.Commands;
using Workflow.Application.BuildFlow.Buttons.DTOs;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.BuildFlow.Buttons.Handlers;
/// <summary>禁用 BuildFlow 按钮定义处理器。用于立即阻止运行时继续展示和执行该按钮。</summary>
public class DisableBuildFlowButtonDefinitionCommandHandler(WorkflowDbContext db)
: IRequestHandler<DisableBuildFlowButtonDefinitionCommand, BuildFlowButtonDefinitionDto>
{
public async Task<BuildFlowButtonDefinitionDto> Handle(
DisableBuildFlowButtonDefinitionCommand request,
CancellationToken ct)
{
var entity = await db.BuildFlowButtons.FindAsync([request.Id], ct)
?? throw new BusinessException($"BuildFlow 按钮 {request.Id} 不存在");
entity.Status = BuildFlowMetadataStatus.Disabled;
await db.SaveChangesAsync(ct);
return ToDto(entity);
}
private static BuildFlowButtonDefinitionDto ToDto(BuildFlowButtonDefinition entity)
=> new(
entity.Id,
entity.ApplicationId,
entity.PageDefinitionId,
entity.FormDefinitionId,
entity.WorkflowDefinitionId,
entity.Code,
entity.Label,
entity.Icon,
entity.Placement,
entity.PermissionCode,
entity.VisibleWhenJson,
entity.DisabledWhenJson,
entity.ConfirmJson,
entity.ActionsJson,
entity.Status,
entity.SortOrder);
}

View File

@ -0,0 +1,44 @@
using MediatR;
using Workflow.Application.BuildFlow.Buttons.Commands;
using Workflow.Application.BuildFlow.Buttons.DTOs;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.BuildFlow.Buttons.Handlers;
/// <summary>发布 BuildFlow 按钮定义处理器。按钮发布后才会被运行时按 Published 状态查询到。</summary>
public class PublishBuildFlowButtonDefinitionCommandHandler(WorkflowDbContext db)
: IRequestHandler<PublishBuildFlowButtonDefinitionCommand, BuildFlowButtonDefinitionDto>
{
public async Task<BuildFlowButtonDefinitionDto> Handle(
PublishBuildFlowButtonDefinitionCommand request,
CancellationToken ct)
{
var entity = await db.BuildFlowButtons.FindAsync([request.Id], ct)
?? throw new BusinessException($"BuildFlow 按钮 {request.Id} 不存在");
entity.Status = BuildFlowMetadataStatus.Published;
await db.SaveChangesAsync(ct);
return ToDto(entity);
}
private static BuildFlowButtonDefinitionDto ToDto(BuildFlowButtonDefinition entity)
=> new(
entity.Id,
entity.ApplicationId,
entity.PageDefinitionId,
entity.FormDefinitionId,
entity.WorkflowDefinitionId,
entity.Code,
entity.Label,
entity.Icon,
entity.Placement,
entity.PermissionCode,
entity.VisibleWhenJson,
entity.DisabledWhenJson,
entity.ConfirmJson,
entity.ActionsJson,
entity.Status,
entity.SortOrder);
}

View File

@ -0,0 +1,12 @@
using MediatR;
using Workflow.Application.Integration.DTOs;
namespace Workflow.Application.Integration.Commands;
/// <summary>
/// 按流程名称触发工作流命令。由 RAG/AI 意图识别后通过 MediatR 调用。
/// </summary>
public sealed record StartWorkflowByNameCommand(
string DefinitionKey,
Guid InitiatorId,
string? Variables) : IRequest<WorkflowTriggerResultDto>;

View File

@ -0,0 +1,10 @@
namespace Workflow.Application.Integration.DTOs;
/// <summary>
/// 工作流触发结果。用于跨模块 CQRS 调用返回 AI/RAG 触发流程的执行结果。
/// </summary>
public sealed record WorkflowTriggerResultDto(
bool Success,
Guid? InstanceId,
string? Title,
string? ErrorMessage);

View File

@ -0,0 +1,10 @@
namespace Workflow.Application.Integration.DTOs;
/// <summary>
/// 可由 AI/RAG 触发的工作流定义摘要。
/// </summary>
public sealed record WorkflowTriggerableDefinitionDto(
Guid Id,
string Key,
string Name,
string? Description);

View File

@ -0,0 +1,24 @@
using MediatR;
using Workflow.Application.Integration.DTOs;
using Workflow.Application.Integration.Queries;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.Integration.Handlers;
/// <summary>
/// 可触发工作流定义查询处理器。供 RAG/AI 在无法匹配流程时展示候选项。
/// </summary>
public sealed class GetTriggerableWorkflowDefinitionsQueryHandler(WorkflowDbContext db)
: IRequestHandler<GetTriggerableWorkflowDefinitionsQuery, List<WorkflowTriggerableDefinitionDto>>
{
public async Task<List<WorkflowTriggerableDefinitionDto>> Handle(
GetTriggerableWorkflowDefinitionsQuery request,
CancellationToken ct)
{
return await db.WorkflowDefinitions
.Where(d => d.Status == DefinitionStatus.Published)
.Select(d => new WorkflowTriggerableDefinitionDto(d.Id, d.Name, d.Name, d.Description))
.ToListAsync(ct);
}
}

View File

@ -0,0 +1,53 @@
using MediatR;
using Workflow.Application.Features.WorkflowInstances.Commands;
using Workflow.Application.Integration.Commands;
using Workflow.Application.Integration.DTOs;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.Integration.Handlers;
/// <summary>
/// 按名称触发工作流命令处理器。负责解析已发布流程定义并复用启动流程实例命令。
/// </summary>
public sealed class StartWorkflowByNameCommandHandler(
WorkflowDbContext db,
ISender sender) : IRequestHandler<StartWorkflowByNameCommand, WorkflowTriggerResultDto>
{
public async Task<WorkflowTriggerResultDto> Handle(
StartWorkflowByNameCommand request,
CancellationToken ct)
{
var definition = await db.WorkflowDefinitions
.FirstOrDefaultAsync(
d => d.Name.Contains(request.DefinitionKey) &&
d.Status == DefinitionStatus.Published,
ct);
if (definition is null)
{
return new WorkflowTriggerResultDto(
false,
null,
null,
$"未找到名为「{request.DefinitionKey}」的已发布流程定义");
}
try
{
var instanceId = await sender.Send(
new StartWorkflowInstanceCommand(
definition.Code ?? definition.Id.ToString(),
$"{request.DefinitionKey}-{DateTime.UtcNow:MMddHHmm}",
request.Variables,
null),
ct);
return new WorkflowTriggerResultDto(true, instanceId, request.DefinitionKey, null);
}
catch (Exception ex)
{
return new WorkflowTriggerResultDto(false, null, null, ex.Message);
}
}
}

View File

@ -0,0 +1,10 @@
using MediatR;
using Workflow.Application.Integration.DTOs;
namespace Workflow.Application.Integration.Queries;
/// <summary>
/// 查询可由 AI/RAG 触发的已发布工作流定义。
/// </summary>
public sealed record GetTriggerableWorkflowDefinitionsQuery
: IRequest<List<WorkflowTriggerableDefinitionDto>>;

View File

@ -1,50 +0,0 @@
using BuildFlow.Shared.Integration;
using Workflow.Application.Features.WorkflowInstances.Commands;
using Workflow.Domain.Enums;
using Workflow.Infrastructure.Persistence;
namespace Workflow.Application.Integration;
/// <summary>
/// 工作流触发服务实现(AI 对话 → 触发工作流)。
/// 串联 RAG ↔ Workflow:AI 识别流程意图后调用,按名称查找定义并启动实例。
/// </summary>
public class WorkflowTriggerService(
WorkflowDbContext db,
MediatR.IMediator mediator) : IWorkflowTriggerService
{
public async Task<WorkflowTriggerResult> TryStartAsync(string definitionKey, Guid initiatorId, string? variables, CancellationToken ct)
{
var definition = await db.WorkflowDefinitions
.FirstOrDefaultAsync(d => d.Name.Contains(definitionKey) && d.Status == DefinitionStatus.Published, ct);
if (definition is null)
{
return new WorkflowTriggerResult(false, null, null, $"未找到名为「{definitionKey}」的已发布流程定义");
}
try
{
var instanceId = await mediator.Send(
new StartWorkflowInstanceCommand(
DefinitionCode: definition.Code ?? definition.Id.ToString(),
Title: $"{definitionKey}-{DateTime.Now:MMddHHmm}",
Variables: variables,
FormDataJson: null), ct);
return new WorkflowTriggerResult(true, instanceId, definitionKey, null);
}
catch (Exception ex)
{
return new WorkflowTriggerResult(false, null, null, ex.Message);
}
}
public async Task<List<WorkflowDefinitionSummary>> GetTriggerableDefinitionsAsync(CancellationToken ct)
{
return await db.WorkflowDefinitions
.Where(d => d.Status == DefinitionStatus.Published)
.Select(d => new WorkflowDefinitionSummary(d.Id, d.Name, d.Name, d.Description))
.ToListAsync(ct);
}
}

View File

@ -1,24 +0,0 @@
namespace BuildFlow.Shared.Integration;
/// <summary>
/// 工作流触发服务接口(跨模块集成:AI 对话 → 触发工作流)。
/// RAG 模块的 AI bot 识别到流程意图后调此接口,Workflow 模块实现它创建流程实例。
/// 放在 Shared 层实现解耦——RAG 不引用 Workflow.Application。
/// </summary>
public interface IWorkflowTriggerService
{
/// <summary>
/// 按定义 key 启动工作流实例。
/// </summary>
/// <param name="definitionKey">流程定义的 Key/名称(如"报销审批"、"请假流程")</param>
/// <param name="initiatorId">发起人 userId</param>
/// <param name="variables">初始变量(JSON 字符串,可含表单数据)</param>
/// <param name="ct">取消令牌</param>
/// <returns>启动结果:实例 ID + 流程标题,失败返回 null + 错误消息</returns>
Task<WorkflowTriggerResult> TryStartAsync(string definitionKey, Guid initiatorId, string? variables, CancellationToken ct);
/// <summary>
/// 查询当前用户可触发的流程定义列表(供 AI bot 展示可选流程)。
/// </summary>
Task<List<WorkflowDefinitionSummary>> GetTriggerableDefinitionsAsync(CancellationToken ct);
}

View File

@ -1,4 +0,0 @@
namespace BuildFlow.Shared.Integration;
/// <summary>流程定义摘要(供 AI 展示)。</summary>
public record WorkflowDefinitionSummary(Guid Id, string Key, string Name, string? Description);

View File

@ -1,4 +0,0 @@
namespace BuildFlow.Shared.Integration;
/// <summary>工作流启动结果。</summary>
public record WorkflowTriggerResult(bool Success, Guid? InstanceId, string? Title, string? ErrorMessage);

View File

@ -90,6 +90,79 @@ public class ChatHubMessagingTests : IAsyncLifetime
}
}
[Fact]
public async Task SendMessage_file_message_preserves_metadata_reply_and_updates_conversation_summary()
{
// Arrange:前端文件发送链路会先完成分片上传,再通过 ChatHub 发送文件名、大小和预览 URL。
_fixture.UserInfoService.SetUser(ImServerFixture.AliceId, "Alice");
_fixture.UserInfoService.SetUser(ImServerFixture.BobId, "Bob");
var convId = await ImTestSeeder.SeedPrivateConversationAsync(
_fixture,
ImServerFixture.AliceId,
ImServerFixture.BobId);
var quoted = new MessageDocument
{
ConversationId = convId,
SenderId = ImServerFixture.BobId,
SenderName = "Bob",
ContentType = MessageContentType.Text,
Content = "请把方案文件发我",
CreatedAt = DateTime.UtcNow.AddSeconds(-10),
};
await DB.Default.SaveAsync(quoted);
var alice = await ConnectAsync(ImServerFixture.AliceId, "Alice");
var bob = await ConnectAsync(ImServerFixture.BobId, "Bob");
MessageDto? bobReceived = null;
bob.On<MessageDto>("OnMessageReceived", m => bobReceived = m);
var metadata = new MessageMetadataDto(
FileName: "BuildFlow-方案.pdf",
FileSize: 2048,
FileUrl: "/api/files/preview?objectKey=multipart%2Fhash",
ImageUrl: null,
ImageWidth: null,
ImageHeight: null,
EmojiCode: null);
// Act:Alice 发送文件消息,并引用 Bob 的原始消息。
await alice.InvokeAsync("SendMessage", new SendMessageRequest(
convId,
MessageContentType.File,
"BuildFlow-方案.pdf",
metadata,
quoted.ID));
await WaitForAsync(() => bobReceived is not null);
// Assert ①:广播 DTO 保留文件元数据和引用消息 ID,前端可以直接渲染文件卡片与引用预览。
Assert.NotNull(bobReceived);
Assert.Equal(MessageContentType.File, bobReceived!.ContentType);
Assert.Equal("BuildFlow-方案.pdf", bobReceived.Content);
Assert.Equal(quoted.ID, bobReceived.ReplyToId);
Assert.Equal("BuildFlow-方案.pdf", bobReceived.Metadata?.FileName);
Assert.Equal(2048, bobReceived.Metadata?.FileSize);
Assert.Equal("/api/files/preview?objectKey=multipart%2Fhash", bobReceived.Metadata?.FileUrl);
// Assert ②:MongoDB 持久化同样保留元数据和引用关系,支撑历史搜索/定位后的再次渲染。
var persisted = await DB.Default.Find<MessageDocument>()
.Match(m => m.ID == bobReceived.Id)
.ExecuteFirstAsync();
Assert.NotNull(persisted);
Assert.Equal(quoted.ID, persisted!.ReplyToId);
Assert.Equal("BuildFlow-方案.pdf", persisted.Metadata?.FileName);
Assert.Equal("/api/files/preview?objectKey=multipart%2Fhash", persisted.Metadata?.FileUrl);
// Assert ③:会话列表摘要使用文件占位,避免长 URL 或元数据污染列表。
await using var scope = _fixture.Services.CreateAsyncScope();
var db = scope.ServiceProvider.GetRequiredService<ImDbContext>();
var conv = await db.Conversations.AsNoTracking().FirstAsync(c => c.Id == convId);
Assert.Equal($"[文件] {metadata.FileName}", conv.LastMessageContent);
Assert.Equal(bobReceived.Id, conv.LastMessageId);
}
[Fact]
public async Task SendMessage_increments_redis_unread_for_other_participant_only()
{

View File

@ -0,0 +1,30 @@
using BuildFlow.Infrastructure.AI;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// AI 对话输出清洗测试。确保模型模板开场不进入最终消息气泡。
/// </summary>
public class AiChatOutputSanitizerTests
{
[Fact]
public void SanitizeFinalTextShouldRemoveRagAssistantBoilerplate()
{
const string raw = "您好!作为RAG知识库助手,我需要您提供具体的问题以及相关的上下文或知识库内容,才能为您准确解答。";
var cleaned = AiChatOutputSanitizer.SanitizeFinalText(raw);
Assert.Equal("请先输入要查询的问题,或选择知识库后再提问。", cleaned);
}
[Fact]
public void SanitizeFinalTextShouldRemoveThinkTags()
{
const string raw = "<think>内部推理</think>\n答案正文";
var cleaned = AiChatOutputSanitizer.SanitizeFinalText(raw);
Assert.Equal("答案正文", cleaned);
}
}

View File

@ -0,0 +1,133 @@
using BuildFlow.Domain.Entities;
using BuildFlow.Domain.Enums;
using BuildFlow.Infrastructure.AI;
using BuildFlow.Infrastructure.Persistence;
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Options;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// AI 运行时模型配置解析测试。确保 Chat/Embedding 调用优先使用模型管理里配置的默认本地模型。
/// </summary>
public class AiModelRuntimeConfigResolverTests
{
[Fact]
public async Task ResolveAsyncShouldReplaceLegacyRagOnlyChatInstructions()
{
var services = new ServiceCollection();
var dbName = $"rag-chat-model-{Guid.NewGuid():N}";
services.AddDbContext<RagDbContext>(options =>
options.UseInMemoryDatabase(dbName));
services.AddScoped<AiModelRuntimeConfigResolver>();
services.AddSingleton(Options.Create(new AiOptions
{
BaseUrl = "http://localhost:1234/v1",
ApiKey = "lm-studio",
ChatModel = "fallback-chat",
EmbeddingModel = "fallback-embedding",
DefaultInstructions = "你是 BuildFlow AI 助手。没有知识库上下文时按普通对话直接回答。",
}));
await using var provider = services.BuildServiceProvider();
await using (var scope = provider.CreateAsyncScope())
{
var db = scope.ServiceProvider.GetRequiredService<RagDbContext>();
await db.AiModels.AddAsync(new AiModel
{
Id = Guid.NewGuid(),
Name = "旧对话模型",
Provider = ModelProvider.LMStudio,
ModelId = "qwen/qwen3.6-27b",
BaseUrl = "http://localhost:1234/v1",
ApiKey = "lm-studio",
ModelType = ModelType.Chat,
IsActive = true,
IsDefault = true,
DefaultInstructions = "你是一个RAG知识库助手,基于提供的上下文准确回答用户问题。如果上下文中没有相关信息,请如实说明。",
CreatedBy = "test",
UpdatedBy = "test",
OperatorIP = "127.0.0.1",
});
await db.SaveChangesAsync();
}
await using (var scope = provider.CreateAsyncScope())
{
var resolver = scope.ServiceProvider.GetRequiredService<AiModelRuntimeConfigResolver>();
var config = await resolver.ResolveAsync(ModelType.Chat, CancellationToken.None);
Assert.Equal("你是 BuildFlow AI 助手。没有知识库上下文时按普通对话直接回答。", config.DefaultInstructions);
}
}
[Fact]
public async Task ResolveAsyncShouldPreferDefaultEmbeddingModelFromDatabase()
{
var services = new ServiceCollection();
var dbName = $"rag-ai-model-{Guid.NewGuid():N}";
services.AddDbContext<RagDbContext>(options =>
options.UseInMemoryDatabase(dbName));
services.AddScoped<AiModelRuntimeConfigResolver>();
services.AddSingleton(Options.Create(new AiOptions
{
BaseUrl = "https://fallback.example/v1",
ApiKey = "fallback-key",
ChatModel = "fallback-chat",
EmbeddingModel = "fallback-embedding",
DefaultInstructions = "fallback instructions",
}));
await using var provider = services.BuildServiceProvider();
await using (var scope = provider.CreateAsyncScope())
{
var db = scope.ServiceProvider.GetRequiredService<RagDbContext>();
await db.AiModels.AddRangeAsync(
new AiModel
{
Id = Guid.NewGuid(),
Name = "旧嵌入模型",
Provider = ModelProvider.OpenAI,
ModelId = "remote-embedding",
BaseUrl = "https://remote.example/v1",
ApiKey = "remote-key",
ModelType = ModelType.Embedding,
IsActive = true,
IsDefault = false,
CreatedBy = "test",
UpdatedBy = "test",
OperatorIP = "127.0.0.1",
},
new AiModel
{
Id = Guid.NewGuid(),
Name = "本地嵌入模型",
Provider = ModelProvider.LMStudio,
ModelId = "text-embedding-embeddinggemma-300m",
BaseUrl = "http://localhost:1234/v1",
ApiKey = "lm-studio",
ModelType = ModelType.Embedding,
IsActive = true,
IsDefault = true,
CreatedBy = "test",
UpdatedBy = "test",
OperatorIP = "127.0.0.1",
});
await db.SaveChangesAsync();
}
await using (var scope = provider.CreateAsyncScope())
{
var resolver = scope.ServiceProvider.GetRequiredService<AiModelRuntimeConfigResolver>();
var config = await resolver.ResolveAsync(ModelType.Embedding, CancellationToken.None);
Assert.Equal("text-embedding-embeddinggemma-300m", config.ModelId);
Assert.Equal("http://localhost:1234/v1", config.BaseUrl);
Assert.Equal("lm-studio", config.ApiKey);
}
}
}

View File

@ -0,0 +1,33 @@
using System.Text.Json;
using BuildFlow.Infrastructure.AI;
using Microsoft.Extensions.AI;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// OpenAI 兼容请求载荷测试。确保本地推理模型默认关闭 reasoning 空转阶段。
/// </summary>
public class OpenAiCompatibleChatRequestPayloadTests
{
[Fact]
public void BuildJsonShouldDisableReasoningEffort()
{
var config = new AiModelRuntimeConfig(
"qwen/qwen3.6-27b",
"http://localhost:1234/v1",
"lm-studio",
null,
128,
0);
var json = OpenAiCompatibleChatRequestPayload.BuildJson(
config,
[new ChatMessage(ChatRole.User, "你好")],
"直接回答。");
using var document = JsonDocument.Parse(json);
Assert.Equal("none", document.RootElement.GetProperty("reasoning_effort").GetString());
}
}

View File

@ -0,0 +1,42 @@
using BuildFlow.Infrastructure.AI;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// OpenAI 兼容流式响应解析测试。覆盖 LM Studio 返回的 content 与 reasoning_content 分离场景。
/// </summary>
public class OpenAiCompatibleChatStreamParserTests
{
[Fact]
public void ReadContentDeltaShouldReturnOnlyVisibleContent()
{
const string data = """
{"choices":[{"delta":{"content":"你好"}}]}
""";
var chunk = OpenAiCompatibleChatStreamParser.ReadContentDelta(data);
Assert.Equal("你好", chunk);
}
[Fact]
public void ReadContentDeltaShouldIgnoreReasoningContent()
{
const string data = """
{"choices":[{"delta":{"reasoning_content":"这里是推理过程"}}]}
""";
var chunk = OpenAiCompatibleChatStreamParser.ReadContentDelta(data);
Assert.Null(chunk);
}
[Fact]
public void ReadContentDeltaShouldIgnoreDoneSentinel()
{
var chunk = OpenAiCompatibleChatStreamParser.ReadContentDelta("[DONE]");
Assert.Null(chunk);
}
}

View File

@ -0,0 +1,115 @@
using BuildFlow.Application.Document.Handlers;
using BuildFlow.Application.Document.Queries;
using BuildFlow.Domain.Entities;
using BuildFlow.Domain.Enums;
using BuildFlow.Domain.Interfaces;
using BuildFlow.Infrastructure.AI;
using BuildFlow.Infrastructure.Persistence;
using Microsoft.AspNetCore.Http;
using Microsoft.EntityFrameworkCore;
using Xunit;
namespace BuildFlow.Rag.Tests;
/// <summary>
/// 文档切片预览测试。预览只提取并分块,不生成 embedding,也不写入正式分块表。
/// </summary>
public class PreviewDocumentChunksQueryHandlerTests
{
[Fact]
public async Task HandleShouldReturnDryRunChunksWithoutPersistingDocumentChunks()
{
var options = new DbContextOptionsBuilder<RagDbContext>()
.UseInMemoryDatabase($"rag-preview-{Guid.NewGuid():N}")
.Options;
await using var db = new RagDbContext(options);
var kbId = Guid.NewGuid();
var docId = Guid.NewGuid();
var filePath = Path.Combine(Path.GetTempPath(), $"preview-{Guid.NewGuid():N}.md");
await File.WriteAllTextAsync(filePath, "# 第一章\n\nBuildFlow 文档切片预览。\n\n# 第二章\n\n上传后先看切片,再手动处理向量。");
try
{
await db.KnowledgeBases.AddAsync(new KnowledgeBase
{
Id = kbId,
Name = "预览知识库",
ChunkSize = 40,
ChunkOverlap = 5,
ChunkingMode = ChunkingMode.Heading,
EmbeddingModel = "text-embedding-embeddinggemma-300m",
Status = KnowledgeBaseStatus.Active,
CreatedBy = "test",
UpdatedBy = "test",
OperatorIP = "127.0.0.1",
});
await db.Documents.AddAsync(new Document
{
Id = docId,
KnowledgeBaseId = kbId,
Title = "preview.md",
FileName = "preview.md",
FilePath = filePath,
FileSize = 64,
ContentType = "text/markdown",
Status = DocumentStatus.Pending,
CreatedBy = "test",
UpdatedBy = "test",
OperatorIP = "127.0.0.1",
});
await db.SaveChangesAsync();
var handler = new PreviewDocumentChunksQueryHandler(
db,
new StubDocumentExtractorFactory(),
new TextChunker(),
new StubFileStorageClient(),
new HttpContextAccessor());
var result = await handler.Handle(new PreviewDocumentChunksQuery(docId), CancellationToken.None);
Assert.Equal(2, result.Count);
Assert.All(result, chunk => Assert.False(chunk.HasEmbedding));
Assert.Contains("第一章", result[0].Content);
Assert.Contains("第二章", result[1].Content);
Assert.Equal(0, await db.DocumentChunks.CountAsync());
}
finally
{
File.Delete(filePath);
}
}
private sealed class StubDocumentExtractorFactory : IDocumentExtractorFactory
{
public IDocumentExtractor GetExtractor(string fileNameOrExtension) => new StubDocumentExtractor();
}
private sealed class StubDocumentExtractor : IDocumentExtractor
{
public IReadOnlyCollection<string> SupportedExtensions { get; init; } = ["md"];
public async Task<ExtractedDocument> ExtractAsync(string filePath, CancellationToken ct = default)
{
return new ExtractedDocument
{
Content = await File.ReadAllTextAsync(filePath, ct),
Title = Path.GetFileName(filePath),
};
}
}
private sealed class StubFileStorageClient : IFileStorageClient
{
public Task UploadAsync(
string bucket,
string objectKey,
Stream content,
string contentType,
string? authToken,
CancellationToken ct = default) => Task.CompletedTask;
public Task<Stream> DownloadAsync(string bucket, string objectKey, string? authToken, CancellationToken ct = default)
=> throw new InvalidOperationException("本用例不应下载文件服务对象");
}
}

View File

@ -49,6 +49,45 @@ public class BuildFlowButtonActionTests
button.Status.Should().Be(BuildFlowMetadataStatus.Draft);
}
[Fact]
public async Task PublishButtonDefinition_ShouldMakeButtonExecutableByRuntime()
{
await using var db = TestDbFactory.Create();
var button = CreateButton(Guid.NewGuid());
button.Status = BuildFlowMetadataStatus.Draft;
db.BuildFlowButtons.Add(button);
await db.SaveChangesAsync();
var handler = new PublishBuildFlowButtonDefinitionCommandHandler(db);
var result = await handler.Handle(
new PublishBuildFlowButtonDefinitionCommand(button.Id),
CancellationToken.None);
result.Status.Should().Be(BuildFlowMetadataStatus.Published);
(await db.BuildFlowButtons.FindAsync(button.Id))!.Status
.Should().Be(BuildFlowMetadataStatus.Published);
}
[Fact]
public async Task DisableButtonDefinition_ShouldRemoveButtonFromRuntime()
{
await using var db = TestDbFactory.Create();
var button = CreateButton(Guid.NewGuid());
db.BuildFlowButtons.Add(button);
await db.SaveChangesAsync();
var handler = new DisableBuildFlowButtonDefinitionCommandHandler(db);
var result = await handler.Handle(
new DisableBuildFlowButtonDefinitionCommand(button.Id),
CancellationToken.None);
result.Status.Should().Be(BuildFlowMetadataStatus.Disabled);
(await db.BuildFlowButtons.FindAsync(button.Id))!.Status
.Should().Be(BuildFlowMetadataStatus.Disabled);
}
[Fact]
public async Task ExecuteAsync_RunFunctionAction_ReturnsFunctionResultAndWritesLog()
{

View File

@ -0,0 +1,113 @@
using FluentAssertions;
using MediatR;
using Workflow.Application.Integration.Commands;
using Workflow.Application.Integration.Handlers;
using Workflow.Application.Integration.Queries;
using Workflow.Application.Features.WorkflowInstances.Commands;
using Workflow.Domain.Entities;
using Workflow.Domain.Enums;
using Xunit;
namespace BuildFlow.Workflow.Tests.Integration;
/// <summary>
/// AI/RAG 触发工作流的 CQRS 契约测试,禁止回退到 shared service bridge。
/// </summary>
public class WorkflowTriggerCqrsTests
{
[Fact]
public async Task StartWorkflowByNameCommand_ShouldDispatchStartInstanceCommand()
{
await using var db = TestDbFactory.Create();
var definition = new WorkflowDefinition
{
Id = Guid.NewGuid(),
Name = "报销审批",
Code = "expense-approval",
Status = DefinitionStatus.Published,
Version = 1,
IsEnabled = true,
};
db.WorkflowDefinitions.Add(definition);
await db.SaveChangesAsync();
var instanceId = Guid.NewGuid();
var sender = new TestSender(instanceId);
var handler = new StartWorkflowByNameCommandHandler(db, sender);
var result = await handler.Handle(
new StartWorkflowByNameCommand("报销", Guid.NewGuid(), """{"amount":1200}"""),
CancellationToken.None);
result.Success.Should().BeTrue();
result.InstanceId.Should().Be(instanceId);
result.Title.Should().Be("报销");
sender.StartCommand.Should().NotBeNull();
sender.StartCommand!.DefinitionCode.Should().Be("expense-approval");
sender.StartCommand.Title.Should().StartWith("报销-");
sender.StartCommand.Variables.Should().Be("""{"amount":1200}""");
}
[Fact]
public async Task GetTriggerableWorkflowDefinitionsQuery_ShouldReturnPublishedDefinitions()
{
await using var db = TestDbFactory.Create();
db.WorkflowDefinitions.AddRange(
new WorkflowDefinition
{
Id = Guid.NewGuid(),
Name = "报销审批",
Code = "expense-approval",
Status = DefinitionStatus.Published,
Version = 1,
IsEnabled = true,
},
new WorkflowDefinition
{
Id = Guid.NewGuid(),
Name = "草稿流程",
Code = "draft-flow",
Status = DefinitionStatus.Draft,
Version = 1,
IsEnabled = true,
});
await db.SaveChangesAsync();
var handler = new GetTriggerableWorkflowDefinitionsQueryHandler(db);
var result = await handler.Handle(new GetTriggerableWorkflowDefinitionsQuery(), CancellationToken.None);
result.Should().ContainSingle();
result[0].Key.Should().Be("报销审批");
result[0].Name.Should().Be("报销审批");
}
private sealed class TestSender(Guid instanceId) : ISender
{
public StartWorkflowInstanceCommand? StartCommand { get; private set; }
public Task<TResponse> Send<TResponse>(IRequest<TResponse> request, CancellationToken cancellationToken = default)
{
if (request is StartWorkflowInstanceCommand command && typeof(TResponse) == typeof(Guid))
{
StartCommand = command;
return Task.FromResult((TResponse)(object)instanceId);
}
throw new NotSupportedException($"Unexpected request {request.GetType().Name}");
}
public Task Send<TRequest>(TRequest request, CancellationToken cancellationToken = default)
where TRequest : IRequest
=> throw new NotSupportedException($"Unexpected request {typeof(TRequest).Name}");
public Task<object?> Send(object request, CancellationToken cancellationToken = default)
=> throw new NotSupportedException($"Unexpected request {request.GetType().Name}");
public IAsyncEnumerable<TResponse> CreateStream<TResponse>(
IStreamRequest<TResponse> request,
CancellationToken cancellationToken = default)
=> throw new NotSupportedException($"Unexpected stream request {request.GetType().Name}");
public IAsyncEnumerable<object?> CreateStream(object request, CancellationToken cancellationToken = default)
=> throw new NotSupportedException($"Unexpected stream request {request.GetType().Name}");
}
}