API集成最佳实践：专业的接入之道

高效的流处理

// 流式响应处理器
class StreamHandler {
  async *handleStream(
    stream: ReadableStream,
    options: StreamOptions = {}
  ): AsyncGenerator<StreamChunk> {
    const reader = stream.getReader();
    const decoder = new TextDecoder();
    let buffer = '';
    
    try {
      while (true) {
        const { done, value } = await reader.read();
        
        if (done) {
          // 处理剩余数据
          if (buffer) {
            yield this.parseChunk(buffer);
          }
          break;
        }
        
        // 解码数据
        buffer += decoder.decode(value, { stream: true });
        
        // 按行分割
        const lines = buffer.split('\n');
        buffer = lines.pop() || '';
        
        for (const line of lines) {
          if (line.trim() === '') continue;
          
          if (line.startsWith('data: ')) {
            const data = line.slice(6);
            
            if (data === '[DONE]') {
              return;
            }
            
            try {
              const chunk = JSON.parse(data);
              
              // 处理数据
              const processed = await this.processChunk(chunk);
              
              // 发送处理后的数据
              yield processed;
              
              // 回调处理
              if (options.onChunk) {
                options.onChunk(processed);
              }
            } catch (error) {
              console.error('Failed to parse chunk:', error);
              
              if (options.onError) {
                options.onError(error);
              }
            }
          }
        }
      }
    } finally {
      reader.releaseLock();
    }
  }
  
  private processChunk(chunk: any): StreamChunk {
    return {
      id: chunk.id,
      content: chunk.choices?.[0]?.delta?.content || '',
      role: chunk.choices?.[0]?.delta?.role,
      finishReason: chunk.choices?.[0]?.finish_reason,
      usage: chunk.usage
    };
  }
  
  // 流聚合器
  async aggregateStream(
    stream: AsyncGenerator<StreamChunk>
  ): Promise<CompleteResponse> {
    let content = '';
    let role = '';
    let usage = null;
    
    for await (const chunk of stream) {
      if (chunk.content) {
        content += chunk.content;
      }
      if (chunk.role) {
        role = chunk.role;
      }
      if (chunk.usage) {
        usage = chunk.usage;
      }
    }
    
    return {
      content,
      role,
      usage
    };
  }
}

// 使用示例
const streamHandler = new StreamHandler();

// React组件中使用
function ChatComponent() {
  const [response, setResponse] = useState('');
  
  const handleStream = async (prompt: string) => {
    const stream = await api.createChatStream(prompt);
    
    for await (const chunk of streamHandler.handleStream(stream, {
      onChunk: (chunk) => {
        setResponse(prev => prev + chunk.content);
      },
      onError: (error) => {
        console.error('Stream error:', error);
      }
    })) {
      // 可以在这里做额外处理
    }
  };
}