最近在重构一个知识库问答系统用户反馈最多的不是答案质量而是AI 回答要等十几秒才出现。这个问题让我意识到在 .NET 技术栈里做 AI 应用流式输出已经不是可选优化项而是基础体验。我把 AgentFramework 封装成的 Agent 交互逻辑、SignalR 的长连接推送通道串起来最终实现了像 ChatGPT 那样逐字渲染的效果。这篇文章把完整链路、协议设计、代码实现和踩过的坑一次性说清楚适合正在做 .NET AI 应用、或者准备把大模型能力接入现有系统的同学参考。1. 为什么非流式输出会让 AI 应用未战先败1.1 大模型生成的本质是 Token 级串行计算大模型回答问题时并不是一次性生成整段文字而是一个 token 一个 token 地推理出来。以 GPT 类模型为例一次完整回答可能包含数百上千个 token模型需要逐个预测下一个 token 的概率分布这个过程是严格串行的。这就带来一个关键指标——TTFTTime To First Token首个 Token 到达时间。非流式请求下服务器要等所有 token 全部生成完毕才一次性把完整答复返回给客户端。假设一个回答有 1500 个 token模型平均每秒生成 50 个 token用户就要在白屏状态下等待 30 秒。哪怕模型只用 10 秒生成完网络传输这 10KB 文本也需要额外时间。流式输出Streaming Output的思路则是模型每生成一个 token服务器立刻把增量推送给客户端。用户看到的不是空白等待而是文字一个接一个打出来。TTFT 被压缩到 2-3 秒剩下的时间虽然总耗时相同但用户的体感完全不同——等待变成了阅读。1.2 很多人只做了传输层没有做渲染层我第一次做流式输出时以为把 HTTP 响应改成Stream就结束了结果前端依然是等全部收完再一次性渲染白屏问题根本没有解决。后来才意识到流式输出必须拆成两个层次传输层服务器到客户端的增量数据传输通道比如 HTTP SSE、WebSocket、SignalR。渲染层前端对增量数据的实时解析与展示包括 Markdown 增量渲染、代码块高亮、光标跟随等。传输层保证数据流起来渲染层保证界面动起来。只看传输层用户依然感觉卡顿只看渲染层没有稳定的推送通道增量数据频繁断连体验同样糟糕。两个层次必须同时做对才算真正的流式输出。2. AgentFramework 与 SignalR 的选型逻辑2.1 AgentFramework 在 .NET 生态里的角色定位直接调用 OpenAI SDK 也能跑通大模型对话但真实业务里的需求往往更复杂要管理多轮上下文、要动态调用工具Function Calling、要支持多个 Agent 协同完成任务。如果把这些逻辑全部写在 Controller 里代码很快会变成一坨无法维护的意大利面。AgentFramework 解决的问题就是把这层交互逻辑抽象成 Agent 模型。它提供了一套面向 Agent 的设计约定Agent代表一个具备特定职责的智能体封装系统提示词、模型配置、工具能力。ChatClient负责与大模型通信的客户端抽象支持同步与异步两种调用方式。AgentTask把一次用户请求包装成任务对象包含输入、上下文、取消令牌等。我实际项目里的初始化代码大致是下面这样using AgentFramework; var builder WebApplication.CreateBuilder(args); // 读取配置中的模型信息 var modelConfig builder.Configuration.GetSection(OpenAI) .GetModelOptions(); // 注册 Agent 核心服务 builder.Services.AddAgentFramework(options { options.DefaultModel modelConfig.Model; options.ApiKey modelConfig.ApiKey; options.Endpoint modelConfig.Endpoint; }); // 注册一个自定义的 QA Agent builder.Services.AddAgentQAAgent(qa);使用的时候只需要把IAgentService注入到业务层就能拿到 Agent 并完成对话public class QAContext { private readonly IAgentService _agentService; public QAContext(IAgentService agentService) { _agentService agentService; } public async TaskIAsyncEnumerablestring AskAsync(string question, CancellationToken ct) { var agent _agentService.GetAgent(qa); return agent.ChatAsync(new AgentTask(question), ct); } }AgentFramework 最大的价值不是省掉那几行 API 调用代码而是把模型配置、上下文管理、工具调用、流式返回这些跨项目复用的逻辑沉淀到了统一抽象层。后续换模型供应商只需要改配置不需要改业务代码。2.2 为什么推送通道选 SignalR 而不是裸 SSE大模型流式输出的本质是单向的服务器不断向客户端推送 token。最简单的实现方式是 HTTP SSEServer-Sent Events一个长连接单向推送代码也最简单。但我最终选了 SignalR原因是业务场景远不止单向推送这么简单。AI Agent 应用中用户随时可能觉得回答跑偏想中止生成也可能需要发送重新生成指令还有可能需要在多个端网页、小程序之间同步会话状态。这些需求要求通道具备双向通信能力而 SSE 只能单向。我把两个方案列了一张对比表对比维度SignalR裸 SSE通信方向双向通信支持客户端向服务器发消息单向只能服务器推给客户端自动重连内置自动重连支持断线恢复需要自己实现重连逻辑协议协商自动协商 WebSocket / SSE / 长轮询固定 HTTP负载均衡扩展支持 Redis Backplane 跨节点广播需要额外方案客户端生态.NET / JS / Java / Python 官方客户端浏览器原生有限支持学习成本需要理解 Hub 模型极低SignalR 在 ASP.NET Core 里有官方原生支持和 AgentFramework 同属 .NET 生态部署时不用额外引入中间件。它内部会优先尝试 WebSocket不支持 WebSocket 的环境下自动降级到 SSE 或长轮询这让我在对接各种客户网络环境时省了很多力气。有一种说法是SignalR 太重AI 推送用 SSE 就够了。这个观点我部分同意如果你的场景只做纯展示没有交互SSE 确实更轻。但 AI Agent 应用几乎都必须支持取消生成和上下文追加这两个操作反过来也要走同一个通道用 SignalR 能保持连接模型的统一。3. 核心链路拆解Hub 设计、消息协议与 Agent 流式读取3.1 Hub 方法设计与连接生命周期SignalR 的调用单元是 Hub。我设计了一个ChatHub对外暴露两个核心方法AskAgent和StopAgent。前者发起一轮问答后者主动中止当前生成。using Microsoft.AspNetCore.SignalR; public class ChatHub : Hub { private readonly IAgentService _agentService; private readonly ILoggerChatHub _logger; public ChatHub(IAgentService agentService, ILoggerChatHub logger) { _agentService agentService; _logger logger; } public override async Task OnConnectedAsync() { // 从 QueryString 或 Token 中解析用户身份绑定到当前 Connection var userId Context.UserIdentifier; await Groups.AddToGroupAsync(Context.ConnectionId, userId); await base.OnConnectedAsync(); } public override async Task OnDisconnectedAsync(Exception? exception) { var userId Context.UserIdentifier; await Groups.RemoveFromGroupAsync(Context.ConnectionId, userId); if (exception ! null) { _logger.LogWarning(exception, Connection {ConnectionId} disconnected with exception, Context.ConnectionId); } await base.OnDisconnectedAsync(exception); } public async Task AskAgent(AgentRequest request) { var connectionId Context.ConnectionId; var ct Context.ConnectionAborted; try { var agent _agentService.GetAgent(request.AgentName); var tokenStream agent.ChatAsync(new AgentTask(request.Question), ct); await foreach (var chunk in tokenStream) { await Clients.Client(connectionId).SendAsync(ReceiveToken, new StreamPayload { Type token, Content chunk, ConversationId request.ConversationId }, ct); } await Clients.Client(connectionId).SendAsync(ReceiveToken, new StreamPayload { Type completed, ConversationId request.ConversationId }, ct); } catch (OperationCanceledException) { // 客户端主动断开或 abort静默处理即可 } catch (Exception ex) { _logger.LogError(ex, Agent invocation failed for conversation {ConversationId}, request.ConversationId); await Clients.Client(connectionId).SendAsync(ReceiveToken, new StreamPayload { Type error, Content ex.Message, ConversationId request.ConversationId }, ct); } } public Task StopAgent(string conversationId) { // 具体中止逻辑见第 5 章 return Task.CompletedTask; } }这段代码有几个容易被忽略的点第一Context.ConnectionAborted是 SignalR 内置的取消令牌客户端断开连接时会自动触发。用它作为 Agent 调用的取消信号可以避免前端页面关闭后后端还在傻傻地生成。第二SendAsync的最后一个参数也传了ct。这个细节很重要因为推送目标已经断开时SendAsync本身也可能抛出异常带上取消令牌可以让它及时停止。3.2 消息协议设计不只推 Token还要推状态刚开始做流式输出我以为只需要把模型的文本增量发送给前端就完事了。实际对接后才发现前端需要区分多种事件文字增量、完整答复、错误信息、会话元数据。如果全部混在一个回调里前端代码会变得非常难写。我设计了一个统一的StreamPayload消息模型消息类型说明触发时机token文本增量片段模型每个 chunk 到达时metadata会话元数据如模型名称、token 用量流开始时发送一次completed流结束信号携带完整答复文本所有 token 发送完毕后error错误信息携带错误描述异常发生时aborted请求已被取消主动中止或超时触发对应的 C# 类型定义public class StreamPayload { public string Type { get; set; } token; public string? Content { get; set; } public string ConversationId { get; set; } string.Empty; public Dictionarystring, object? Metadata { get; set; } }为什么要设计completed事件并携带完整文本因为前端增量渲染 Markdown 时会面临标签被截断的问题——比如模型一次返回**加粗**但**可能被拆到两个 chunk 里。如果只靠增量片段做最终渲染格式必然错乱。所以我的策略是流过程中用轻量方式展示增量流结束时用完整文本做一次终版渲染。这样既有实时打字效果又能保证最终展示质量。3.3 多 Agent 并行时消息顺序是个大问题实际业务里我遇到过需要两个 Agent 并行回答的场景一个负责找资料一个负责写初稿。两个 Agent 各自返回IAsyncEnumerablestring如果简单地Task.WhenAll两个循环它们的输出会互相穿插前端看到的文字完全错乱。我用Channel做了输出汇聚把每个 Agent 的输出按时间顺序写入一个统一通道再由一个独立的推送任务从通道读取并发送给客户端public async IAsyncEnumerablestring MergeAgentStreams( IEnumerableIAsyncEnumerablestring streams, CancellationToken ct) { var channel Channel.CreateUnboundedstring( new UnboundedChannelOptions { SingleReader true }); var writerTasks streams.Select(stream WriteToChannelAsync(stream, channel.Writer, ct)).ToArray(); // 后台写任务统一启动 _ Task.Run(async () { await Task.WhenAll(writerTasks); channel.Writer.TryComplete(); }, ct); await foreach (var item in channel.Reader.ReadAllAsync(ct)) { yield return item; } }这么做的好处是消息写入顺序完全由模型输出的实际节奏决定没有人为的时间片分配前端渲染顺序是确定的。但要注意Channel.Unbounded在高并发下可能存在内存压力实际项目中我建议根据模型输出速率设置一个合理的容量上限。4. 前端实时渲染Vue 3 SignalR 客户端 Markdown 增量渲染4.1 建立 SignalR 连接与消息接收前端我用的 Vue 3 TypeScriptSignalR 官方提供microsoft/signalr客户端库通过 npm 直接安装npm install microsoft/signalr连接代码封装成一个独立的 composable方便多个组件复用import * as signalR from microsoft/signalr; let connection: signalR.HubConnection | null null; export function useChatHub() { async function connect(): PromisesignalR.HubConnection { if (connection connection.state Connected) { return connection; } connection new signalR.HubConnectionBuilder() .withUrl(/hubs/chat) .withAutomaticReconnect([0, 1000, 5000, 10000]) .configureLogging(signalR.LogLevel.Warning) .build(); await connection.start(); return connection; } function onToken(callback: (payload: StreamPayload) void) { if (!connection) return; connection.on(ReceiveToken, callback); } async function askAgent(conversationId: string, question: string) { if (!connection) return; await connection.invoke(AskAgent, { conversationId, question, agentName: qa, }); } async function stopAgent(conversationId: string) { if (!connection) return; await connection.invoke(StopAgent, conversationId); } return { connect, onToken, askAgent, stopAgent }; }withAutomaticReconnect我设置了渐进式重连间隔0 毫秒、1 秒、5 秒、10 秒。这样做的好处是短暂的网络抖动可以无感恢复不需要用户手动刷新。但要注意重连成功后 SignalR 会恢复连接但不会自动恢复 Hub 方法调用所以我写在onreconnected事件里把当前未完成的会话重新发起一遍。4.2 Markdown 增量渲染的标签截断处理这是整个流式输出方案里我踩得最深的一个坑。模型输出的 chunk 是随机的一个**加粗**可能被切成**加和**粗**两段一个代码块可能先收到 csharp 再收到后面的代码内容。直接用 marked 或 markdown-it 渲染增量会出现两个恶心的问题一是短标签闪烁页面上一会儿显示**加一会儿变成完整的**加粗**视觉上非常跳跃。 二是渲染器报错部分渲染器遇到未闭合的 HTML 标签会直接抛异常甚至把整个页面搞崩。我的解决方案分为三个阶段阶段一流进行中不直接渲染 Markdown而是把增量纯文本追加到消息对象里配合一个模拟打字的效果。页面看到的是正在打出的文字遇到代码块或特殊格式时也不做高亮处理。阶段二completed 事件到达用完整文本走一次完整的 Markdown 渲染并开启代码高亮。这是最终展示给用户的稳定版本。阶段三渲染校正如果 Markdown 中有表格、代码块等复杂结构在终版渲染完成后做一次 DOM 检查确保没有残留的半截标签。实际代码里我会把两个版本的文本都存起来interface ChatStreamState { conversationId: string; streamingText: string; // 流式过程中的纯文本增量 finalText: string; // completed 事件中的完整文本 status: streaming | completed | error | aborted; }渲染时优先展示finalText如果还没有finalText就展示streamingText并在末尾加一个光标闪烁的 CSS 动画模拟打字效果。4.3 网络层的err_incomplete_chunked_encoding问题开发联调时我遇到过一个典型的浏览器报错net::ERR_INCOMPLETE_CHUNKED_ENCODING。这个问题意味着服务端声明的流式响应chunked transfer encoding在传输途中被截断浏览器没有收到完整的结束标记。排查下来原因有三个第一个是代理缓冲。开发环境没有 nginx但部署环境在 nginx 后面nginx 默认会缓冲上游响应导致浏览器一直等不到数据或者等 buffer 满了才一次性推送。解决办法是在 nginx 配置里关闭对 AI 接口的缓冲location /hubs/ { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; }第二个是应用层异常中断。后端在流式输出过程中抛出异常但异常信息没有被正确捕获连接被直接断开浏览器就会收到一个不完整的 chunked 响应。我给所有 Agent 调用入口都加了 try-catch确保异常时至少发送一个error事件并且让连接正常关闭。第三个是超时配置过短。大模型生成长回答可能需要很久默认的 60 秒超时在生成 2000 token 时经常被打断。我把 nginx 的proxy_read_timeout调到了 300 秒同时在前端 SignalR 配置里把serverTimeoutInMilliseconds适当调大。5. Abort 机制取消请求是让流式输出真正可用的关键5.1 CancellationToken 的传递链路很多刚做 AI 应用的同学会忽略取消机制觉得回答生成完就完事了。但真实场景里用户随时可能发现问题想停止生成如果后端不响应取消请求模型会继续推理到最后一个 token——每一秒都在烧钱。SignalR 天然提供了一条从客户端到服务端的取消链路客户端调用stopAgent(conversationId)。Hub 的StopAgent方法被触发找到该会话对应的CancellationTokenSource并取消。CancellationToken通过AgentTask传入agent.ChatAsync。Agent 框架内部把该取消令牌传递给底层模型 SDK 的 HTTP 请求。在 Hub 内部维护一个会话与CancellationTokenSource的映射private readonly ConcurrentDictionarystring, CancellationTokenSource _activeTasks new(); public async Task AskAgent(AgentRequest request) { var cts new CancellationTokenSource(); _activeTasks[request.ConversationId] cts; var linkedCts CancellationTokenSource.CreateLinkedTokenSource( Context.ConnectionAborted, cts.Token); try { // ... 遍历 token 流推送数据 } finally { _activeTasks.TryRemove(request.ConversationId, out _); cts.Dispose(); } } public Task StopAgent(string conversationId) { if (_activeTasks.TryRemove(conversationId, out var cts)) { cts.Cancel(); } return Task.CompletedTask; }注意我用了CreateLinkedTokenSource把「客户端断开」和「用户主动中止」两个信号合并成一个令牌。这样无论用户是点取消按钮还是直接关掉浏览器模型推理都能被及时中断。5.2 取消场景下的行为验证我在实测中验证过三类取消行为结果是场景一用户点取消按钮。StopAgent被调用CancellationTokenSource取消Agent 内部的await foreach抛出OperationCanceledException外层 catch 捕获后静默退出不再推送任何数据。前端收到aborted事件把状态置为已中止。场景二用户直接刷新页面。浏览器先断开 WebSocket 连接SignalR 服务端立即触发OnDisconnectedAsyncContext.ConnectionAborted被取消。因为我把这个令牌链接到了 Agent 调用所以模型推理随之停止。场景三服务端主动超时。我在配置里设置单次回答的最大 token 预算和最大时长超过后主动取消CancellationTokenSource防止模型无限生成下去。这个策略对控制成本非常有效。5.3 取消后的异常处理细节取消操作不是无痕的OperationCanceledException只是最常见的异常实际还会遇到底层 http 请求被中断时抛出TaskCanceledException。部分模型 SDK 在超时会抛TimeoutException而不是取消异常。推送目标已断开SendAsync抛出Connection closed异常。我建议在 catch 异常时做一次异常类型编排不要一网打尽也不要只 catch 一种catch (OperationCanceledException) when (ct.IsCancellationRequested) { _logger.LogInformation(Agent stream cancelled for {ConversationId}, conversationId); } catch (TaskCanceledException) when (ct.IsCancellationRequested) { _logger.LogInformation(Agent HTTP request cancelled for {ConversationId}, conversationId); } catch (TimeoutException ex) { _logger.LogWarning(ex, Agent stream timed out for {ConversationId}, conversationId); }判断ct.IsCancellationRequested是为了区分主动取消和意外异常。如果是主动取消日志级别降低不打扰运维如果是超时或真实异常日志级别提高方便追踪。6. 实操中的稳定性优化与踩坑心得6.1 每个 token 都推那不行要加缓冲最朴素的实现是模型返回一个 chunk 就推给前端一个 chunk。但实际模型 API 的 chunk 是突击式的有的 chunk 只包含几个字符甚至有的只包含一个空格。高频小包推送会带来两个问题网络吞吐浪费、前端渲染线程被频繁打断。我的做法是加一层时间片缓冲把 50 毫秒内的所有 chunk 合并成一个批再推给前端。这样既不会改变打字效果50 毫秒人眼感知不到又能显著减少推送次数。public async IAsyncEnumerablestring BufferStreamAsync( IAsyncEnumerablestring source, TimeSpan window, CancellationToken ct) { var buffer new StringBuilder(); var timer System.Threading.Timer.Create(..., dueTime: -1, period: -1); // 简化示意按时间窗口合并积压内容 await foreach (var chunk in source.WithCancellation(ct)) { buffer.Append(chunk); if (buffer.Length 16) // 也按字符数兜底 { yield return buffer.ToString(); buffer.Clear(); } } if (buffer.Length 0) { yield return buffer.ToString(); } }6.2 Agent 工具调用结果与流式文本的顺序问题如果 Agent 需要调用工具比如查数据库、调内部 API模型输出会分成两段第一段是推理过程文本中间夹着工具调用工具返回结果后模型继续输出第二段文本。这种情况下如果不在后端对消息做排序前端会看到第二段推理突然出现在屏幕上。我的处理方案是在 Agent 内部把工具调用结果单独放入一个metadata消息不混入 token 流。前端收到metadata事件后在聊天气泡下方展示已调用工具资料检索耗时 2.3 秒等最终完整文本到达后再渲染成完整回答。这样既保留了工具调用的可解释性又不会打断主文本的流式体验。6.3 多轮上下文的组装不能用简单追加很多人做多轮对话时简单地把历史消息全部拼进 prompt。这在 token 数少的时候问题不大但长会话下有两个问题一是超出模型的上下文窗口二是模型被无关信息干扰回答质量下降。我在 AgentFramework 封装层做了会话管理策略说明适用场景滑动窗口保留最近 N 条消息通用对话关键信息压缩用一小段摘要替换较长的历史内容长文档问答检索增强只取与当前问题相关的历史片段知识库系统实现上没有银弹但默认推荐滑动窗口 摘要压缩的组合效果最稳。6.4 我最后悔没早点知道的一个小技巧SignalR 的完整消息在 Redis 背板下默认会做序列化存储消息体过大会影响性能和内存。我在推送completed事件时携带完整答复文本内容经常达到几千字给 Redis 背板造成了不小的压力。后来我把策略改成completed事件只携带会话 ID 和简短状态完整文本让前端在流式过程中自行拼接保存不再重复传送。这一调整直接让网关层的内存峰值降了一半。如果你的系统还没上 Redis 背板这个优化可以先不做但只要有多实例部署的规划建议提前按这个思路设计消息协议。从 AgentFramework 抽象 Agent 交互到 SignalR 建立双向通道再到前端增量渲染和取消机制这条链路看起来长但每一环都是必须的。我在实际项目中还发现流式输出上线后不仅是体验提升连回答太长干脆不看的反馈都少了——因为用户习惯了边看边读而不需要一直盯着转圈。如果你正在做 .NET AI 的应用建议先把流式链路搭稳再谈 Agent 编排和工具调用这套地基打扎实了上层不管怎么加功能都稳得住。