fix: truely streaming

Author: qddegtya

.gitignore

@@ -60,3 +60,4 @@ temp
 # claude code
 .claude/
 .serena/
+CLAUDE.md

examples/serve-astack/backend/src/routes/chat.ts

@@ -121,6 +121,7 @@ export default async function chatRoutes(fastify: FastifyInstance) {
  reply.header('X-Vercel-AI-Data-Stream', 'v1');
 
  let fullContent = '';
+ let lastSentLength = 0; // 🎯 追踪已发送的内容长度
 
  try {
  // Stream the agent processing using AI SDK Data Stream Protocol
@@ -144,37 +145,38 @@ export default async function chatRoutes(fastify: FastifyInstance) {
 
  case 'assistant_message': {
  if (chunk.content) {
+ // 🎯 真流式修复：现在 chunk.content 已经是实时incremental的了
+ // 不需要二次分割，直接发送即可获得最佳流式体验
+
  fullContent = chunk.content;
 
- // 智能流式传输：针对代码内容优化
- let chunks: string[];
- if (STREAMING_CONFIG.streamByCharacter) {
- // 字符级流式
- chunks = chunk.content.split('');
- } else {
- // 智能分块：对于包含代码的内容，使用更细粒度的分割
- if (
- chunk.content.includes('```') ||
- chunk.content.includes('def ') ||
- chunk.content.includes('function ') ||
- chunk.content.includes('class ')
- ) {
- // 代码内容：按行分割，保持良好的流式效果
- chunks = chunk.content.split(/(\n)/);
- } else {
- // 普通文本：按词语分割
- chunks = chunk.content.split(/(\s+)/);
- }
- }
+ // 获取新增的内容（incremental delta）
+ const newContent = fullContent.slice(lastSentLength);
+ lastSentLength = fullContent.length;
 
- for (const textChunk of chunks) {
- if (textChunk) {
- // 跳过空字符串
- // Text Part: 0:string\n
- const textPart = `0:${JSON.stringify(textChunk)}\n`;
+ if (newContent) {
+ // 🔧 保持配置化的流式传输选项
+ if (STREAMING_CONFIG.streamByCharacter) {
+ // 字符级流式：对新增内容进行字符分割
+ const chars = newContent.split('');
+ for (const char of chars) {
+ if (char) {
+ const textPart = `0:${JSON.stringify(char)}\n`;
+ reply.raw.write(textPart);
+
+ if (STREAMING_CONFIG.delayPerToken > 0) {
+ await new Promise(resolve =>
+ setTimeout(resolve, STREAMING_CONFIG.delayPerToken)
+ );
+ }
+ }
+ }
+ } else {
+ // 直接发送增量内容（推荐，性能最佳）
+ const textPart = `0:${JSON.stringify(newContent)}\n`;
  reply.raw.write(textPart);
 
- // 可配置的延迟
+ // 可选的小延迟（现在主要用于视觉效果）
  if (STREAMING_CONFIG.delayPerToken > 0) {
  await new Promise(resolve =>
  setTimeout(resolve, STREAMING_CONFIG.delayPerToken)

packages/components/package.json

@@ -1,6 +1,6 @@
 {
  "name": "@astack-tech/components",
- "version": "0.1.1-beta.1",
+ "version": "0.1.1-beta.2",
  "description": "Components for the Astack AI Framework.",
  "main": "dist/index.cjs",
  "module": "dist/index.js",

packages/components/src/agents/StreamingAgent.ts

@@ -186,7 +186,13 @@ export class StreamingAgent extends Component {
  const tools = (this.agent as unknown as { tools?: unknown[] }).tools || [];
  const model = (
  this.agent as unknown as {
- model: { chatCompletion: (...args: unknown[]) => Promise<unknown> };
+ model: {
+ chatCompletion: (...args: unknown[]) => Promise<unknown>;
+ streamChatCompletion: (
+ messages: Message[],
+ options?: { temporaryTools?: unknown[] }
+ ) => AsyncGenerator<Partial<MessageWithToolCalls>>;
+ };
  }
  ).model;
  const verbose = (this.agent as unknown as { verbose?: boolean }).verbose || false;
@@ -214,28 +220,54 @@ export class StreamingAgent extends Component {
  type: 'model_thinking',
  };
 
- // 调用模型获取回复（与原 Agent 完全一致）
- const modelResponse = (await model.chatCompletion(currentMessages, {
+ // 调用模型获取流式回复（真正的SSE流式）
+ const streamGenerator = model.streamChatCompletion(currentMessages, {
  temporaryTools: tools,
- })) as MessageWithToolCalls;
+ });
+
+ let accumulatedContent = '';
+ let finalToolCalls: MessageWithToolCalls['tool_calls'] = [];
+ const modelResponse: MessageWithToolCalls = {
+ role: 'assistant',
+ content: '',
+ tool_calls: [],
+ };
+
+ // 处理真正的流式响应
+ for await (const chunk of streamGenerator) {
+ if (chunk.content) {
+ accumulatedContent += chunk.content;
+ modelResponse.content = accumulatedContent;
+
+ if (verbose) {
+ console.log('[StreamingAgent Debug] 收到流式chunk:', chunk.content);
+ }
+
+ // 实时流式输出每个 content chunk
+ yield {
+ type: 'assistant_message',
+ content: accumulatedContent,
+ toolCalls: finalToolCalls,
+ };
+ }
+
+ // 处理工具调用（通常在流式结束时到达）
+ if (chunk.tool_calls) {
+ finalToolCalls = chunk.tool_calls;
+ modelResponse.tool_calls = finalToolCalls;
+ }
+ }
 
  if (verbose) {
  console.log(
- '[StreamingAgent Debug] 收到模型回复:',
+ '[StreamingAgent Debug] 流式完成，最终回复:',
  JSON.stringify(modelResponse, null, 2)
  );
  }
 
  // 保存模型回复作为最后的助手消息
  lastAssistantMessage = modelResponse;
 
- // 流式输出：助手消息
- yield {
- type: 'assistant_message',
- content: modelResponse.content || '',
- toolCalls: modelResponse.tool_calls,
- };
-
  // 检查是否有工具调用（与原 Agent 逻辑一致）
  if (!modelResponse.tool_calls || modelResponse.tool_calls.length === 0) {
  if (verbose) {