import { Body, Controller, Inject, Post, Res, UseGuards } from '@nestjs/common'; import { FastifyReply } from 'fastify'; import { StreamChatRequest } from '@shared/ai-gateway/types/chat.types'; import { ProviderRouterService } from '@shared/ai-gateway/router/provider-router.service'; import { ClientJwtAuthGuard } from '../../auth/client-jwt-auth.guard'; import { ApiBearerAuth, ApiBody, ApiConsumes, ApiOkResponse, ApiOperation, ApiProduces, ApiTags, } from '@nestjs/swagger'; function formatSse(event: string, data: unknown) { return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`; } function sleep(ms: number) { return new Promise((resolve) => setTimeout(resolve, ms)); } @ApiTags('Client Chat') @ApiBearerAuth('access-token') @Controller('client/v1/chat') export class ChatController { constructor( @Inject(ProviderRouterService) private readonly router: ProviderRouterService, ) {} @Post('completions/stream') @UseGuards(ClientJwtAuthGuard) @ApiOperation({ summary: '统一流式 Chat 接口(SSE)' }) @ApiConsumes('application/json') @ApiProduces('text/event-stream') @ApiBody({ schema: { type: 'object', description: '统一 chat 请求体,支持指定平台或自动路由', properties: { model: { type: 'string', description: '模型名(不传时 provider 使用默认模型)', example: 'qwen-plus', }, platform: { type: 'string', description: '目标平台(auto 或不传表示自动路由)', enum: ['auto', 'qwen', 'deepseek', 'volc'], example: 'qwen', }, messages: { type: 'array', description: '聊天消息列表', items: { type: 'object', properties: { role: { type: 'string', description: '消息角色', enum: ['system', 'user', 'assistant'], example: 'user', }, content: { type: 'string', description: '消息文本内容', example: '你是谁', }, }, required: ['role', 'content'], }, }, }, required: ['messages'], }, }) @ApiOkResponse({ description: 'SSE 流式响应:meta -> delta -> usage -> done', schema: { type: 'string', example: 'event: meta\\ndata: {"requestId":"chatcmpl_xxx","platform":"qwen","model":"qwen-plus"}\\n\\n' + 'event: delta\\ndata: {"delta":"你好"}\\n\\n' + 'event: usage\\ndata: {"promptTokens":10,"completionTokens":20,"totalTokens":30}\\n\\n' + 'event: done\\ndata: {"finishReason":"stop"}\\n\\n', }, }) async streamChat( @Body() body: StreamChatRequest, @Res() reply: FastifyReply, ) { const response = await this.router.routeAndStream(body); reply.raw.setHeader('Content-Type', 'text/event-stream; charset=utf-8'); reply.raw.setHeader('Cache-Control', 'no-cache, no-transform'); reply.raw.setHeader('Connection', 'keep-alive'); reply.raw.setHeader('X-Accel-Buffering', 'no'); reply.raw.flushHeaders?.(); reply.raw.write( formatSse('meta', { requestId: response.requestId, platform: response.providerCode, model: response.model, }), ); for (const chunk of response.chunks) { reply.raw.write(formatSse('delta', { delta: chunk.content })); await sleep(120); } reply.raw.write(formatSse('usage', response.usage)); reply.raw.write(formatSse('done', { finishReason: 'stop' })); reply.raw.end(); } }