企业官网建设流程全解析

1. 项目概述：为什么流式响应是 Coding Agent 的“呼吸”

如果你正在构建一个 Coding Agent，或者任何需要与大型语言模型（LLM）进行长时间、多轮次交互的 AI 应用，那么“流式响应”绝对不是你锦上添花的功能，而是决定用户体验成败的“呼吸系统”。想象一下，你向一个助手提问，它需要思考一分钟，然后才把完整的答案一股脑地“吐”给你。在这漫长的等待中，你无法判断它是卡住了、出错了，还是在认真工作。这种体验无疑是糟糕的，尤其是在代码生成、调试、解释等场景下，用户需要实时看到模型的“思考”过程，以便及时引导或纠正。

这就是“流式响应”要解决的核心痛点。它允许服务器将 LLM 生成的内容像水流一样，以数据块（chunk）的形式持续推送给客户端，而不是等待整个内容生成完毕再一次性返回。对于 Coding Agent 而言，这意味着：

• 实时反馈：用户能立即看到模型生成的第一个单词、第一行代码，感知到 Agent 正在工作，降低等待焦虑。
• 动态交互：在代码生成过程中，如果用户发现方向不对，可以随时中断，而不是等一个可能完全错误的冗长结果。
• 性能感知：流式传输能更早地开始渲染内容，从用户感知上大幅缩短响应时间，提升应用响应速度。

本篇文章，我们将深入探讨如何为你的 Coding Agent 实现稳定、高效的流式响应。我们将超越简单的“Hello World”示例，聚焦于生产环境中必须考虑的细节：如何选择协议、如何处理网络中断、如何设计前后端协作、以及如何优化用户体验。无论你使用的是 OpenAI 的 GPT 系列、开源的 Llama 或 Qwen，还是其他任何提供流式接口的模型，这里讨论的原理和模式都是相通的。

2. 技术选型：SSE 与 WebSocket 的深度对比

实现流式通信，主流有两种技术：Server-Sent Events (SSE)和WebSocket。很多文章会简单地告诉你“用 SSE 就够了”，但作为开发者，我们必须理解背后的“为什么”，才能做出最适合自己场景的选择。

2.1 SSE：为服务器到客户端的单向流而生

SSE 是一种基于 HTTP 的轻量级协议。它的核心思想是，客户端发起一个普通的 HTTP 请求，但服务器不立即关闭连接，而是保持连接打开，并持续发送一系列格式化的“事件”数据。

工作原理简述：

1. 客户端（浏览器）使用EventSourceAPI 向一个特定端点发起 GET 请求，并在请求头中设置Accept: text/event-stream。
2. 服务器响应时，设置Content-Type: text/event-stream，并保持连接不关闭。
3. 服务器通过这个持久的连接，持续发送遵循特定格式的数据块。每个数据块以data:开头，以两个换行符\n\n结束。例如：

   data: 这是第一段流式内容\n\n data: 这是第二段内容\n\n

4. 客户端EventSource会自动解析这些消息，并触发onmessage事件。

SSE 的优势与局限：

• 优势：
  • 协议简单：基于 HTTP/HTTPS，无需额外的协议升级握手，兼容性极佳，几乎不受防火墙或代理限制。
  • 自动重连：EventSource内置了连接断开后的自动重连机制，对于不稳定的网络环境非常友好。
  • 轻量级：协议开销小，特别适合服务器向客户端推送文本信息的场景，如新闻推送、股票行情、以及我们这里的 LLM 文本流。
• 局限：
  • 单向通信：只能从服务器向客户端推送数据。如果 Coding Agent 需要在生成过程中接收用户的实时干预指令（例如“停，重写这个函数”），SSE 本身无法支持。通常需要配合另一个独立的 HTTP 请求通道来实现。
  • 文本协议：虽然可以传输 JSON 字符串，但原生设计是针对文本的。传输二进制数据（如音频、视频流）不是它的强项。
  • 连接数限制：浏览器对同一个域名下的 HTTP 连接数有上限（通常为6个），一个持久的 SSE 连接会占用其中一个。对于需要大量并发长连接的场景需要谨慎。

2.2 WebSocket：全双工实时通信的瑞士军刀

WebSocket 提供了在单个 TCP 连接上进行全双工通信的能力。连接建立后，客户端和服务器可以随时相互发送数据。

工作原理简述：

1. 客户端发起一个带有Upgrade: websocket头的 HTTP 请求，进行协议升级握手。
2. 握手成功后，连接从 HTTP 协议切换为 WebSocket 协议。
3. 此后，双方可以通过send方法发送数据，并通过onmessage事件接收数据，数据格式可以是文本或二进制。

WebSocket 的优势与局限：

• 优势：
  • 全双工：双向实时通信是它的核心优势。对于需要复杂交互的 Coding Agent（例如，用户一边看代码生成，一边可以发送“解释这行”、“重构变量名”等指令），WebSocket 是更自然的选择。
  • 低延迟：建立连接后，数据传输头部开销极小，延迟非常低。
  • 二进制支持：原生支持二进制帧，适合传输多种类型的数据。
• 局限：
  • 协议更复杂：需要处理握手、帧解析、心跳保活等，实现复杂度高于 SSE。
  • 无自动重连：连接断开后，需要开发者自己实现重连逻辑。
  • 可能遇到代理问题：某些古老的或配置严格的代理服务器可能不支持 WebSocket 协议升级。

2.3 决策指南：为你的 Coding Agent 选择什么？

我的经验与建议： 对于大多数初、中阶的 Coding Agent 项目，我强烈建议从 SSE 开始。原因如下：

1. 场景匹配：LLM 生成代码的过程，本质上是服务器将生成的 token 流式推送给客户端，这是一个典型的单向推送场景。SSE 就是为此而生的。
2. 简单可靠：基于 HTTP，意味着你可以复用现有的认证、负载均衡、监控体系。EventSource的自动重连能省去大量边缘情况处理代码。
3. 快速上手：你可以在一个下午就搭建出可用的流式响应原型，把精力集中在 Prompt 工程和 Agent 逻辑上，而不是通信协议上。

当你需要实现以下功能时，才需要考虑升级到 WebSocket：

• 实时双向对话：用户可以在 Agent 生成代码时，随时插入评论或指令。
• 多模态流：需要同时流式传输代码和生成的图表、解释音频等。
• 超高并发与低延迟：对延迟有极致要求，且能驾驭 WebSocket 的复杂性和状态管理。

在本文的后续实现中，我们将以SSE作为主要技术栈进行详解，因为它是最贴合“让 LLM 流式响应”这一核心需求的、性价比最高的方案。

3. 后端实现：构建健壮的流式 API 端点

后端是流式响应的发动机。我们的目标是构建一个能够稳定、高效地从 LLM 获取流式数据，并将其规范化为 SSE 格式推送给前端的服务。这里以 Python 的 FastAPI 框架和 OpenAI 兼容的 API 为例，但原理适用于任何语言和模型。

3.1 依赖安装与基础设置

首先，确保你的环境已安装必要的库。我们将使用openai库（或兼容的客户端，如litellm）来调用模型，使用sse-starlette或sse_starlette来方便地生成 SSE 响应。

pip install fastapi uvicorn openai sse-starlette

3.2 核心 API 端点实现

我们将创建一个/stream的 POST 端点，它接收用户的请求（如代码生成任务描述），然后流式返回模型的响应。

from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import openai import asyncio import json from sse_starlette.sse import EventSourceResponse app = FastAPI() # 配置你的 LLM 客户端，这里以 OpenAI 格式为例 # 实际可能是 OpenAI, Azure OpenAI, 或本地部署的 vLLM 等服务 client = openai.AsyncOpenAI( api_key="your-api-key", base_url="https://api.openai.com/v1" # 或你的本地/第三方端点 ) async def generate_streaming_response(prompt: str, model: str = "gpt-4"): """ 核心生成器函数：从 LLM 获取流式响应，并转换为 SSE 格式。 """ try: # 调用 LLM 的流式接口 stream = await client.chat.completions.create( model=model, messages=[{"role": "user", "content": prompt}], stream=True, # 关键参数：启用流式输出 temperature=0.7, max_tokens=2000, ) # 异步迭代流式响应 async for chunk in stream: # 提取 delta 中的内容 content = chunk.choices[0].delta.content if content is not None: # 将内容封装为 SSE 事件数据 # 我们发送一个 JSON 字符串，包含内容和可能的元数据（如是否结束） event_data = json.dumps({ "content": content, "finished": False }) yield event_data # 流结束时，发送一个结束标记事件 yield json.dumps({"content": "", "finished": True}) except Exception as e: # 发生错误时，发送错误信息并标记结束 error_data = json.dumps({ "content": f"\n\n[流式生成发生错误: {str(e)}]", "finished": True, "error": True }) yield error_data @app.post("/api/stream-code") async def stream_code(request: Request): """ 流式代码生成端点。 期望的请求体 JSON: {"prompt": "用户的任务描述", "model": "可选，模型名称"} """ data = await request.json() prompt = data.get("prompt", "") model = data.get("model", "gpt-4") if not prompt: # 对于错误请求，也可以返回一个立即结束的流 async def error_stream(): yield json.dumps({"content": "错误：请求中未提供有效的 prompt。", "finished": True, "error": True}) return EventSourceResponse(error_stream()) # 使用 EventSourceResponse 包装我们的生成器 # 它会自动设置正确的 Content-Type: text/event-stream return EventSourceResponse( generate_streaming_response(prompt, model), ping=15000 # 每15秒发送一个“ping”注释以保持连接活跃，防止超时 )

代码关键点解析：

1. stream=True：这是调用 LLM API 时开启流式响应的关键参数。没有它，你会一直等待整个响应完成。
2. 异步生成器 (async for)：我们使用async for来异步地消费 LLM 返回的流。这对于处理高并发请求至关重要，它不会阻塞事件循环。
3. SSE 数据格式：我们每次yield一个 JSON 字符串。前端需要解析这个 JSON 来获取content和状态标记（finished,error）。这种结构比只发送纯文本更灵活，便于扩展（例如未来加入思考过程、工具调用等元数据）。
4. EventSourceResponse：来自sse-starlette，它帮我们处理了 SSE 协议细节，包括自动添加data:前缀和双换行符，以及可选的ping机制来保持连接。
5. 错误处理：在try...except中包裹核心逻辑，确保即使 LLM API 调用失败，也能向客户端发送一个有意义的错误消息并正常结束流，而不是直接断开连接导致前端无法感知。

3.3 生产环境增强考虑

上面的代码是一个可运行的原型，但要用于生产，还需要考虑以下几点：

• 超时与中断处理：用户可能中途关闭页面或取消请求。后端需要监听连接断开事件，并立即终止昂贵的 LLM 生成过程，以节省资源。在 FastAPI 中，你可以检查request.is_disconnected()（在生成器内部定期检查比较麻烦，通常需要结合背景任务和取消令牌）。
• 速率限制与缓存：为 SSE 连接实现速率限制，防止滥用。对于相同的提示，可以考虑缓存首个 chunk 或完整响应，但要注意流式场景下缓存的新鲜度。
• 结构化数据与工具调用：如果你的 Coding Agent 使用 ReAct 模式或 Function Calling，流式响应可能需要传输更复杂的结构化数据（如[THOUGHT]...[/THOUGHT]），而不仅仅是纯文本。需要在数据协议设计时预留字段。
• 使用更通用的客户端：考虑使用litellm这样的库，它统一了 OpenAI、Anthropic、Cohere、本地模型等众多接口，让你的后端代码更容易切换模型提供商。

4. 前端实现：优雅地消费与展示流式内容

前端的目标是创建一个流畅、用户友好的界面，实时接收并渲染从后端 SSE 端点推送来的代码片段。我们将使用现代 JavaScript（或 TypeScript）和EventSourceAPI。

4.1 基础 EventSource 连接

首先，我们创建一个函数来建立 SSE 连接并处理数据。

class StreamingCodeAgent { constructor(apiEndpoint = '/api/stream-code') { this.apiEndpoint = apiEndpoint; this.eventSource = null; this.onDataCallback = null; this.onFinishCallback = null; this.onErrorCallback = null; } /** * 开始流式代码生成 * @param {string} prompt - 用户输入的提示词 * @param {string} model - 选择的模型 */ startStreaming(prompt, model = 'gpt-4') { // 先关闭可能存在的旧连接 this.close(); // 构建请求体 const requestBody = { prompt, model }; // 使用 URLSearchParams 或直接发送 JSON，注意后端接收方式 // 这里我们使用 POST 并发送 JSON，但 EventSource 原生只支持 GET。 // 因此需要一个变通方案： // 方案A: 后端改为 GET，参数放查询字符串（有长度限制）。 // 方案B: 使用 Fetch API 的流式响应，放弃 EventSource。 // 方案C: 先 POST 创建一个会话，返回一个带 token 的 GET 流式端点。 // 这里演示方案B，因为它更灵活且支持POST。 this.useFetchStreaming(prompt, model); } /** * 使用 Fetch API 实现流式读取（推荐，支持POST） */ async useFetchStreaming(prompt, model) { try { const response = await fetch(this.apiEndpoint, { method: 'POST', headers: { 'Content-Type': 'application/json', }, body: JSON.stringify({ prompt, model }), }); if (!response.ok || !response.body) { throw new Error(`HTTP error! status: ${response.status}`); } // 获取可读流 const reader = response.body.getReader(); const decoder = new TextDecoder('utf-8'); let buffer = ''; while (true) { const { done, value } = await reader.read(); if (done) { // 流完全结束 if (this.onFinishCallback) this.onFinishCallback(); break; } // 解码 chunk 并添加到缓冲区 buffer += decoder.decode(value, { stream: true }); // 解析缓冲区中的完整 SSE 行（以 \n\n 分隔） const lines = buffer.split('\n\n'); // 最后一行可能是不完整的，保留在缓冲区 buffer = lines.pop() || ''; for (const line of lines) { if (line.startsWith('data: ')) { const dataStr = line.slice(6).trim(); // 去掉 'data: ' if (dataStr) { try { const data = JSON.parse(dataStr); // 调用回调函数处理数据 if (data.error && this.onErrorCallback) { this.onErrorCallback(data.content); } else if (this.onDataCallback) { this.onDataCallback(data.content, data.finished); } // 如果收到结束信号，可以跳出循环（但让外层循环自然结束更安全） if (data.finished) { reader.cancel(); // 可选：提前取消读取 if (this.onFinishCallback) this.onFinishCallback(); return; } } catch (e) { console.error('Failed to parse SSE data:', e, 'Raw:', dataStr); } } } // 忽略 'event:', 'id:', 'retry:' 等其他 SSE 字段，或处理 ping 注释 } } } catch (error) { console.error('Streaming failed:', error); if (this.onErrorCallback) this.onErrorCallback(`连接失败: ${error.message}`); } } // 注册回调函数 onData(callback) { this.onDataCallback = callback; return this; // 支持链式调用 } onFinish(callback) { this.onFinishCallback = callback; return this; } onError(callback) { this.onErrorCallback = callback; return this; } // 关闭连接 close() { if (this.eventSource) { this.eventSource.close(); this.eventSource = null; } } }

为什么不用原生的EventSource？原生EventSource不支持 POST 请求和自定义请求头，这在传递较长的 prompt 或需要认证时很不方便。上面的useFetchStreaming方法使用 Fetch API 手动处理 SSE 流，虽然代码量稍多，但获得了完全的灵活性。

4.2 与 UI 框架集成（以 React 为例）

现在，我们将这个流式客户端集成到一个 React 组件中。

import React, { useState, useRef, useEffect } from 'react'; function CodeGenerationView() { const [inputPrompt, setInputPrompt] = useState('请用Python写一个快速排序函数，并添加详细注释。'); const [isStreaming, setIsStreaming] = useState(false); const [generatedCode, setGeneratedCode] = useState(''); const [statusMessage, setStatusMessage] = useState(''); const streamAgentRef = useRef(null); useEffect(() => { // 初始化 Agent 实例 streamAgentRef.current = new StreamingCodeAgent('http://your-backend.com/api/stream-code'); // 设置回调 streamAgentRef.current .onData((chunk, isFinished) => { // 收到数据块，追加到生成的代码中 setGeneratedCode(prev => prev + chunk); if (isFinished) { setIsStreaming(false); setStatusMessage('生成完成！'); } }) .onFinish(() => { setIsStreaming(false); setStatusMessage('流已结束。'); }) .onError((errorMsg) => { setIsStreaming(false); setStatusMessage(`错误: ${errorMsg}`); setGeneratedCode(prev => prev + `\n\n--- 生成中断: ${errorMsg} ---`); }); // 组件卸载时清理 return () => { if (streamAgentRef.current) { streamAgentRef.current.close(); } }; }, []); const handleGenerate = () => { if (!inputPrompt.trim() || isStreaming) return; setGeneratedCode(''); // 清空之前的内容 setStatusMessage('正在生成代码...'); setIsStreaming(true); // 开始流式生成 streamAgentRef.current.startStreaming(inputPrompt); }; const handleStop = () => { if (streamAgentRef.current) { streamAgentRef.current.close(); setIsStreaming(false); setStatusMessage('已手动停止。'); } }; return ( <div className="code-gen-container"> <h2>智能代码生成助手</h2> <div className="input-area"> <textarea value={inputPrompt} onChange={(e) => setInputPrompt(e.target.value)} placeholder="描述你想要的代码功能..." rows="4" disabled={isStreaming} /> <div className="button-group"> <button onClick={handleGenerate} disabled={isStreaming || !inputPrompt.trim()}> {isStreaming ? '生成中...' : '开始生成'} </button> {isStreaming && ( <button onClick={handleStop} className="stop-button"> 停止生成 </button> )} </div> <div className="status">{statusMessage}</div> </div> <div className="output-area"> <label>生成的代码：</label> <pre className="code-block"> <code>{generatedCode || '// 代码将在这里实时显示...'}</code> </pre> {/* 可以在这里添加一个代码编辑器组件（如 Monaco Editor）来获得高亮和编辑功能 */} </div> </div> ); } export default CodeGenerationView;

4.3 用户体验优化技巧

1. 打字机效果：与其一次性追加整个 chunk，可以模拟打字效果，逐个字符添加到 DOM。但这会增加前端复杂度，对于长代码可能影响性能。一个折中方案是“行级”或“词级”流式渲染。
2. 自动滚动：当代码不断追加时，自动将滚动条保持在底部，让用户始终看到最新内容。可以在onData回调中操作 DOM 元素的scrollTop属性。
3. 语法高亮：流式过程中实时进行语法高亮是个挑战，因为代码不完整。可以：
   • 使用能处理不完整语法的前端高亮库（如 Shiki、Prism.js 的某些插件）。
   • 在高亮前对不完整的行或语法结构进行简单清理或占位。
   • 更简单的做法：等流式结束后再进行一次完整的高亮。
4. 加载指示器：在代码开始生成前，可以显示一个闪烁的光标或“正在思考...”的动画，给予用户即时反馈。
5. 错误状态与重试：当流式中断或出错时，除了显示错误信息，还应提供一个“重试”按钮，让用户可以重新发送请求，而不是重新输入所有内容。

5. 高级话题与生产环境挑战

将流式响应投入生产环境，意味着要面对更复杂的网络环境、更高的性能要求和更严苛的稳定性需求。

5.1 连接稳定性与重连策略

网络是不稳定的。SSE 连接可能因网络波动、代理超时、服务器重启而中断。

• 后端保活（Ping）：我们在后端代码中设置了ping=15000，这会在连接空闲时定期发送冒号开头的注释行（:\n\n），以保持 TCP 连接活跃，防止被中间设备（如负载均衡器、代理服务器）因超时而关闭。这个时间间隔需要根据你的基础设施调整。
• 前端自动重连：如果使用原生EventSource，它内置了重连逻辑。如果使用我们基于 Fetch 的实现，需要自己实现。一个简单的策略是：在onError或连接异常关闭后，等待一个指数退避的时间（如 1秒，2秒，4秒...），然后重新调用startStreaming。注意，对于用户主动停止的情况，不应触发自动重连。

5.2 流式传输中的上下文管理

对于多轮对话的 Coding Agent，你需要维护对话历史。在流式场景下，这带来两个问题：

1. 本次流式响应中引用历史：这通常由后端处理。你需要将完整的对话历史（包括本次用户提问）作为 messages 列表发送给 LLM API。
2. 将本次流式结果加入历史，供下一轮使用：前端需要在流式完全结束后，将最终生成的完整内容（generatedCode）发送回后端，由后端存储到对话会话中。切勿在流式过程中每收到一个 chunk 就更新服务器端的历史记录，这会产生大量无效请求。

5.3 性能优化：从 Token 到屏幕

• Chunk 合并与节流：LLM API 可能以极快的速度返回 token（每个 token 可能就是一个字符）。如果每收到一个 token 就更新一次 React 状态（setGeneratedCode），会导致界面频繁重绘，性能低下。一个常见的优化是“缓冲”：在前端累积一小段时间（如 50-100ms）或一定数量字符（如 20个字符）的 chunk，然后再批量更新状态。这能显著提升渲染性能。
• 使用 Web Worker：对于非常密集的流式更新和语法高亮计算，可以考虑将流式数据的处理和组装放到 Web Worker 中，避免阻塞主线程，保持 UI 响应流畅。
• 后端响应优化：确保你的后端服务器（如 Uvicorn/Gunicorn）配置了合适的异步工作模式和超时设置，能够高效处理大量并发的长连接（SSE 连接）。

5.4 安全与监控

• 认证与授权：SSE 端点也需要保护。你可以在请求头中携带 Token（如 JWT）。对于基于 Fetch 的实现，这很容易。对于希望用原生EventSource的情况，可能需要通过 URL 查询参数传递 Token（注意安全风险），或者先通过一个普通 API 认证，获取一个短期有效的、专用于 SSE 连接的令牌。
• 速率限制：根据用户 ID 或 IP 对/stream端点进行速率限制，防止恶意用户耗尽你的 LLM API 额度或服务器资源。
• 监控与日志：记录流式请求的开始、结束、持续时间、消耗的 token 数。这对于成本核算、性能分析和故障排查至关重要。注意，日志本身不应阻塞流式响应。

6. 常见问题排查与实战技巧

在实际开发中，你一定会遇到各种“坑”。这里记录了一些典型问题及其解决方案。

6.1 连接立即关闭或收不到数据

• 检查响应头：后端必须设置Content-Type: text/event-stream。如果使用了StreamingResponse或EventSourceResponse，它们通常会帮你设置。
• 检查 CORS：如果前端与后端域名不同，后端必须正确配置 CORS，允许前端域名的请求，并暴露必要的头（如Content-Type）。
• 检查网络代理和防火墙：某些企业网络环境可能会拦截或篡改长连接。尝试在简单网络环境下测试。确保你的负载均衡器（如 Nginx）配置了合适的超时时间（例如proxy_read_timeout 300s;）。
• 查看浏览器开发者工具：在 Network 标签页查看对你的 SSE 端点的请求。状态应该是 “Pending” 或 “200”，并且类型是 “eventsource”。点击请求，在 “Response” 或 “EventStream” 标签页查看是否有数据流进来。

6.2 数据流中断或不完整

• 后端生成器提前退出：确保你的后端异步生成器函数 (generate_streaming_response) 中没有未捕获的异常，并且yield语句被执行。在finally块中打印日志有助于调试。
• 前端缓冲区解析错误：我们手动解析 SSE 格式时，如果数据块 (chunk) 的边界恰好切在\n\n中间，会导致解析错误。代码中的buffer机制就是为了处理这种不完整帧。确保你的解析逻辑足够健壮。
• Nginx/Apache 缓冲：反向代理服务器默认可能会缓冲上游（你的后端应用）的响应，直到达到一定大小或超时后才发送给客户端。这会导致流式响应失去“实时性”。你需要在代理配置中禁用缓冲：

  # Nginx 配置示例 location /api/stream-code { proxy_pass http://your_backend; proxy_set_header Connection ''; proxy_http_version 1.1; chunked_transfer_encoding off; # 对于 SSE，有时需要关闭分块编码 proxy_buffering off; # 关键：关闭代理缓冲 proxy_cache off; # 关闭缓存 proxy_read_timeout 300s; # 设置长的读取超时 }

6.3 流式内容格式混乱

• LLM 输出格式控制：LLM 可能在流式输出中插入 Markdown 代码块标记（如```python）。如果你的前端只是简单显示文本，这些标记会显得很乱。你需要在 Prompt 中明确要求模型“直接输出纯代码，不要包含任何 Markdown 标记”，或者在前后端对输出进行后处理，过滤掉这些标记。
• 处理换行符：LLM 输出的换行符 (\n) 在 HTML 中默认不显示为换行。你需要用 CSSwhite-space: pre-wrap;或white-space: pre;来保留格式，或者将\n替换为<br />。

6.4 内存与资源泄漏

• 及时关闭连接：在 React 组件卸载、用户离开页面或主动停止时，务必调用agent.close()或reader.cancel()来主动关闭连接和释放资源。
• 后端连接管理：对于每个活跃的 SSE 连接，后端都会保持一个协程或线程。确保在连接断开时（通过捕获GeneratorExit异常或检查客户端断开），及时清理资源并终止 LLM 的生成调用（如果 API 支持的话）。

一个关键的实操心得：在开发初期，不要过度优化。先让最基本的流式功能跑通（后端能 yield 数据，前端能收到并显示）。然后，逐步添加错误处理、连接管理、UI 优化。分阶段迭代，会让你更容易定位问题。流式响应的核心价值在于“实时性”和“可中断性”，只要实现了这两点，你的 Coding Agent 体验就已经上了一个大台阶。