流式响应处理

ProductReady 的 AI Agent 系统使用流式响应来提供实时的用户体验。本文档介绍如何实现和处理流式响应。

概述

流式响应允许 AI 生成的内容实时传输到客户端，而不是等待整个响应完成后再显示。这提供了更好的用户体验，特别是对于长内容生成任务。

核心特性

实时流式传输：内容逐步生成并立即显示
可恢复流：支持网络中断后恢复流式传输
工具调用支持：流中包含工具调用和执行结果
类型安全：完整的 TypeScript 类型支持

基础用法

1. 服务端实现

所有 Agent Providers 都返回兼容 AI SDK 的流式响应：

import { GaiaAgentProvider } from "agentlib/providers";
import type { CoreMessage } from "ai";

// 创建 Provider
const provider = new GaiaAgentProvider({
  model: "gpt-4",
  apiKey: process.env.OPENAI_API_KEY,
});

// 准备消息
const messages: CoreMessage[] = [
  { role: "user", content: "写一篇关于 AI 的文章" }
];

// 调用流式接口
const result = await provider.stream(messages);

// 转换为 HTTP 响应
const response = result.toUIMessageStreamResponse({
  onFinish: async ({ messages }) => {
    console.log("流式传输完成:", messages);
  },
});

return response;

2. 客户端使用

使用 AI SDK 的 useChat Hook：

'use client';

import { useChat } from 'ai/react';

export function ChatComponent() {
  const { messages, input, handleInputChange, handleSubmit, isLoading } = useChat({
    api: '/api/chat',
  });

  return (
    <div>
      <div>
        {messages.map((message) => (
          <div key={message.id}>
            <strong>{message.role}:</strong>
            <p>{message.content}</p>
          </div>
        ))}
      </div>
      
      <form onSubmit={handleSubmit}>
        <input
          value={input}
          onChange={handleInputChange}
          disabled={isLoading}
          placeholder="输入消息..."
        />
        <button type="submit" disabled={isLoading}>
          发送
        </button>
      </form>
    </div>
  );
}

高级功能

可恢复流（Resumable Streams）

支持网络中断后恢复流式传输，基于 AI SDK v6 的 resumable stream 模式：

import { createConsumeSseStreamCallback, createResumeResponse } from "agentlib/stream";
import { after } from "next/server";

// POST 处理器 - 创建新流
export async function POST(req: Request) {
  const { messages, taskId } = await req.json();
  
  const provider = new GaiaAgentProvider({
    model: "gpt-4",
    apiKey: process.env.OPENAI_API_KEY,
  });
  
  const result = await provider.stream(messages);
  
  return result.toUIMessageStreamResponse({
    onFinish: async ({ messages }) => {
      // 保存消息到数据库
      await saveMessages(taskId, messages);
      // 清除活动流 ID
      await saveActiveStreamId(taskId, null);
    },
    // 配置可恢复流
    consumeSseStream: createConsumeSseStreamCallback({
      waitUntil: after,
      onStreamCreated: async (streamId) => {
        // 保存流 ID 以便后续恢复
        await saveActiveStreamId(taskId, streamId);
      },
    }),
  });
}

// GET 处理器 - 恢复现有流
export async function GET(req: Request) {
  const { searchParams } = new URL(req.url);
  const taskId = searchParams.get('taskId');
  
  if (!taskId) {
    return new Response('Missing taskId', { status: 400 });
  }
  
  // 读取活动流 ID
  const activeStreamId = await readActiveStreamId(taskId);
  
  // 创建恢复响应
  return createResumeResponse({
    activeStreamId,
    waitUntil: after,
    onStreamDone: () => saveActiveStreamId(taskId, null),
  });
}

自定义流 ID 生成

控制消息 ID 的生成方式：

import { createIdGenerator } from "ai";

const result = await provider.stream(messages);

return result.toUIMessageStreamResponse({
  generateMessageId: createIdGenerator({ 
    prefix: "msg", 
    size: 16 
  }),
  onFinish: async ({ messages }) => {
    // 处理完成逻辑
  },
});

工具调用处理

流式响应中包含工具调用信息：

import { tool } from "ai";
import { z } from "zod";

// 定义工具
const tools = {
  generate_article: tool({
    description: "生成文章内容",
    inputSchema: z.object({
      title: z.string(),
      content: z.string(),
    }),
    execute: async ({ title, content }) => {
      // 执行文章生成
      await createArticle(title, content);
      return { success: true };
    },
  }),
};

// 使用带工具的 Provider
const provider = new GaiaAgentProvider({
  model: "gpt-4",
  apiKey: process.env.OPENAI_API_KEY,
  tools,
});

const result = await provider.stream(messages);

return result.toUIMessageStreamResponse({
  onFinish: async ({ messages }) => {
    // 消息中包含工具调用和结果
    messages.forEach(msg => {
      if (msg.toolInvocations) {
        console.log("工具调用:", msg.toolInvocations);
      }
    });
  },
});

客户端高级用法

处理工具调用

'use client';

import { useChat } from 'ai/react';

export function AdvancedChatComponent() {
  const { messages, input, handleInputChange, handleSubmit } = useChat({
    api: '/api/chat',
    onToolCall: ({ toolCall }) => {
      console.log('工具调用:', toolCall);
      // 可以在 UI 中显示工具执行状态
    },
  });

  return (
    <div>
      {messages.map((message) => (
        <div key={message.id}>
          <p>{message.content}</p>
          
          {message.toolInvocations?.map((tool) => (
            <div key={tool.toolCallId}>
              <strong>工具: {tool.toolName}</strong>
              {tool.state === 'result' && (
                <pre>{JSON.stringify(tool.result, null, 2)}</pre>
              )}
            </div>
          ))}
        </div>
      ))}
      
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
        <button type="submit">发送</button>
      </form>
    </div>
  );
}

流式中断与重新连接

'use client';

import { useChat } from 'ai/react';
import { useEffect } from 'react';

export function RobustChatComponent() {
  const { messages, input, handleInputChange, handleSubmit, reload } = useChat({
    api: '/api/chat',
    onError: (error) => {
      console.error('流式错误:', error);
      // 可以自动重试
    },
  });

  // 处理页面恢复
  useEffect(() => {
    const handleVisibilityChange = () => {
      if (document.visibilityState === 'visible') {
        // 页面重新可见时，尝试恢复流
        reload();
      }
    };

    document.addEventListener('visibilitychange', handleVisibilityChange);
    return () => {
      document.removeEventListener('visibilitychange', handleVisibilityChange);
    };
  }, [reload]);

  return (
    <div>
      {messages.map((message) => (
        <div key={message.id}>
          <p>{message.content}</p>
        </div>
      ))}
      
      <form onSubmit={handleSubmit}>
        <input value={input} onChange={handleInputChange} />
        <button type="submit">发送</button>
      </form>
    </div>
  );
}

实际应用示例

完整的 API Route 实现

// app/api/chat/route.ts
import { GaiaAgentProvider } from "agentlib/providers";
import { createConsumeSseStreamCallback, createResumeResponse } from "agentlib/stream";
import { after } from "next/server";
import { auth } from "@clerk/nextjs/server";

// POST - 创建新的流式对话
export async function POST(req: Request) {
  const { userId } = await auth();
  
  if (!userId) {
    return new Response('Unauthorized', { status: 401 });
  }

  const { messages, taskId } = await req.json();

  // 创建 Provider
  const provider = new GaiaAgentProvider({
    model: process.env.AGENT_MODEL || "gpt-4",
    apiKey: process.env.OPENAI_API_KEY,
    tools: getCustomTools(taskId),
  });

  // 清除之前的流
  if (taskId) {
    await saveActiveStreamId(taskId, null);
  }

  // 执行流式调用
  const result = await provider.stream(messages);

  return result.toUIMessageStreamResponse({
    onFinish: async ({ messages }) => {
      // 保存消息
      if (taskId) {
        await saveMessages(taskId, messages);
        await saveActiveStreamId(taskId, null);
      }
      
      // 计费处理
      const assistantMsgs = messages.filter(m => m.role === 'assistant');
      if (assistantMsgs.length > 0) {
        await consumeUsage(userId, 'ai_credits', assistantMsgs.length);
      }
    },
    consumeSseStream: taskId
      ? createConsumeSseStreamCallback({
          waitUntil: after,
          onStreamCreated: (streamId) => saveActiveStreamId(taskId, streamId),
        })
      : undefined,
  });
}

// GET - 恢复现有流
export async function GET(req: Request) {
  const { searchParams } = new URL(req.url);
  const taskId = searchParams.get('taskId');

  if (!taskId) {
    return new Response('Missing taskId', { status: 400 });
  }

  const activeStreamId = await readActiveStreamId(taskId);

  return createResumeResponse({
    activeStreamId,
    waitUntil: after,
    onStreamDone: () => saveActiveStreamId(taskId, null),
  });
}

最佳实践

1. 错误处理

// ✅ 好：正确处理流式错误
try {
  const result = await provider.stream(messages);
  return result.toUIMessageStreamResponse({
    onFinish: async ({ messages }) => {
      await saveMessages(taskId, messages);
    },
  });
} catch (error) {
  console.error("流式调用失败:", error);
  
  // 返回错误响应
  return new Response(
    JSON.stringify({ error: "流式传输失败" }), 
    { 
      status: 500,
      headers: { 'Content-Type': 'application/json' }
    }
  );
}

2. 消息持久化

// ✅ 好：在 onFinish 中保存消息
return result.toUIMessageStreamResponse({
  onFinish: async ({ messages }) => {
    try {
      // 过滤和保存消息
      const validMessages = messages.filter(m => 
        m.content && m.content.length > 0
      );
      await saveMessages(taskId, validMessages);
    } catch (error) {
      console.error("保存消息失败:", error);
      // 不要抛出错误，避免中断流
    }
  },
});

3. 资源清理

// ✅ 好：确保清理流资源
return result.toUIMessageStreamResponse({
  onFinish: async ({ messages }) => {
    await saveMessages(taskId, messages);
    // 清理活动流 ID
    await saveActiveStreamId(taskId, null);
  },
  consumeSseStream: createConsumeSseStreamCallback({
    waitUntil: after,
    onStreamCreated: async (streamId) => {
      await saveActiveStreamId(taskId, streamId);
      
      // 设置超时清理
      setTimeout(async () => {
        const current = await readActiveStreamId(taskId);
        if (current === streamId) {
          await saveActiveStreamId(taskId, null);
        }
      }, 5 * 60 * 1000); // 5 分钟后清理
    },
  }),
});

4. 客户端重连策略

// ✅ 好：实现智能重连
const { messages, input, handleInputChange, handleSubmit, reload } = useChat({
  api: '/api/chat',
  onError: (error) => {
    console.error('流式错误:', error);
    
    // 网络错误时重试
    if (error.message.includes('network') || error.message.includes('fetch')) {
      setTimeout(() => reload(), 1000);
    }
  },
  // 启用自动重试
  maxRetries: 3,
  retryDelay: 1000,
});

性能优化

消息优化

import { optimizeMessagesForContext } from "agentlib/chat";

// 优化消息历史，避免 token 超限
const optimized = optimizeMessagesForContext(messages);
const result = await provider.stream(optimized);

流式缓冲

// 在客户端使用节流减少重渲染
import { useState, useEffect } from 'react';

export function ChatComponent() {
  const { messages } = useChat({ api: '/api/chat' });
  const [displayMessages, setDisplayMessages] = useState(messages);
  
  // 使用节流更新显示的消息
  useEffect(() => {
    const timer = setTimeout(() => {
      setDisplayMessages(messages);
    }, 100);
    
    return () => clearTimeout(timer);
  }, [messages]);

  return (
    <div>
      {displayMessages.map((message) => (
        <div key={message.id}>{message.content}</div>
      ))}
    </div>
  );
}

故障排除

流式中断

问题： 流式传输中途中断

解决方案：

检查网络连接
验证 API 密钥是否有效
检查是否达到速率限制
使用可恢复流模式

消息丢失

问题： 部分消息未保存

解决方案：

确保 onFinish 正确实现
添加错误处理和日志
使用数据库事务保证原子性

工具调用失败

问题： 工具执行异常

解决方案：

验证工具 schema 定义
添加工具执行错误处理
记录工具调用日志

流式响应处理

流式响应处理

概述

核心特性

基础用法

1. 服务端实现

2. 客户端使用

高级功能

可恢复流（Resumable Streams）

自定义流 ID 生成

工具调用处理

客户端高级用法

处理工具调用

流式中断与重新连接

实际应用示例

完整的 API Route 实现

最佳实践

1. 错误处理

2. 消息持久化

3. 资源清理

4. 客户端重连策略

性能优化

消息优化

流式缓冲

故障排除

流式中断

消息丢失

工具调用失败

相关资源

下一步

On this page