响应的格式
data: {"choices":[{"delta":{"content":"秋"}}]} data: {"choices":[{"delta":{"content":"风"}}]} data: {"choices":[{"delta":{"content":"萧瑟"}}]} data: [DONE]
- 帧以
data:开头、空行分隔 - 每帧是一个 JSON 片段（
delta是增量，不是全文） - 最后一帧固定
[DONE]表示结束。注意是增量不是全量—— 客户端要自己把每个 delta 拼起来
消费模型从 “返回值” 变成 “回调”
- 消费模型是指你的程序调用大模型 API 并获取结果的方式
- 返回值是调用函数卡在哪里等函数返回
- 回调就是：你不是等它给你一个返回值，而是你提前告诉它"如果发生了什么事，来调我的某个方法"。
- 就是提前给框架注册三个回调函数，框架在合适的时间来调用
model.chatStream("你好",newStreamCallback(){voidonPartialResponse(Stringchunk){...}// 有新内容了，来调我voidonCompleteResponse(FullResponser){...}// 完整了，来调我voidonError(Exceptione){...}// 出错了，来调我});
流式输出的并发安全问题
- 非流式是拉模式，主动拉结果，等函数返回
- 流式是推模式，调用函数后不阻塞，接着往下，等结果被推回来
- 所以说主流程需要被阻塞等待chat()返回
CountDownLatchlatch=newCountDownLatch(1);model.chat("你好",handler);// 发起请求，立刻返回latch.await();// ← 主线程在这里卡住，等回调说"好了"
- 最后一个回调函数里面有CountDownLatch++，让程序放行
- 回调运行在别的线程上，计数之类的需要用线程安全的工具
转为流式输出的话装饰器怎么办
如果是非流式的话问题很简单
//调用 → 等完整结果 → 拿到返回值 → 后处理ChatResponser=delegate.doChat(request);// 等，拿到完整结果doSomethingWith(r);// 记账、打日志等returnr;// 返回给上层
但是返回时流式的所以复杂了点
- 结果不是"一个东西"，而是一连串回调事件。你没法"拿到返回值再处理"，因为根本没有返回值，只有事件流。所以你必须在中间插一个代理 handler（拦截器），拦截事件、做处理、再转发
谁是上层谁是下层
业务代码（Controller/Service） ← 最上层 ↓ 调用 Cache 装饰器 ← 外层（上层） ↓ 调用 Router 装饰器 ↓ 调用 Metering 装饰器 ↓ 调用 OpenAI 真实模型 ← 最下层（底层）
在此之前先讲一下装饰器是怎么组在一起的
// 原始模型ChatModelopenAiModel=newOpenAiChatModel(apiKey);// 一层一层包上去ChatModelmetered=newMeteringChatModel(openAiModel,meteringService);ChatModelrouted=newRouterChatModel(metered,classifier);ChatModelcached=newCacheChatModel(routed,cacheService);// 业务代码用 cached 就行ChatResponser=cached.doChat(request);
classMeteringChatModelimplementsChatModel{privatefinalChatModeldelegate;// 指向下一个privatefinalMeteringServicemeteringService;MeteringChatModel(ChatModeldelegate,MeteringServicemeteringService){this.delegate=delegate;this.meteringService=meteringService;}@OverridepublicChatResponsedoChat(ChatRequestrequest){ChatResponser=delegate.doChat(request);// 调下一层meteringService.recordUsage(r);// 拦截：记账returnr;// 返回}}
cached.doChat(req) → routed.doChat(req) // Cache 调 Router → metered.doChat(req) // Router 调 Metering → openAi.doChat(req)// Metering 调真实模型 ← 返回 ChatResponse ← 记账后返回 ← 路由后返回 ← 缓存后返回
接下来是流式的组装方式
- 组装链路是一样的，外层包含内层
- 但是装饰器的每一层内部不再是return，而是插入代理handler
- 先看最下层是怎么写接收器的
// 业务代码：我定义一个"接收器"StreamingChatResponseHandlermyHandler=newStreamingChatResponseHandler(){@OverridepublicvoidonPartialResponse(Stringpartial){// OpenAI 每来一块数据，就调我这个方法System.out.print(partial);// 打印到屏幕上，像打字机一样}@OverridepublicvoidonCompleteResponse(ChatResponseresponse){// 全部发完了，调我这个方法System.out.println("完了，总共用了 "+response.getMetadata().getUsage()+" 个token");}@OverridepublicvoidonError(Throwableerror){System.out.println("出错了："+error.getMessage());}};// 业务代码：把"接收器"交给 OpenAI 模型openAiModel.doChat(request,myHandler);
- 这里面的三个接收器是最普通的
- 然后再看看次下层是怎么写接收器的（计量层）
classStreamingMeteringChatModelimplementsStreamingChatModel{privatefinalStreamingChatModeldelegate;// 指向 OpenAI@OverridepublicvoiddoChat(ChatRequestrequest,StreamingChatResponseHandlerhandler){// 注意：这里的 handler 参数就是 myHandler（业务代码传进来的）// 我造一个新的 handler（偷换）StreamingChatResponseHandlerproxyHandler=newStreamingChatResponseHandler(){@OverridepublicvoidonPartialResponse(Stringpartial){// 我什么也不做，直接转交给原来的 handlerhandler.onPartialResponse(partial);}@OverridepublicvoidonCompleteResponse(ChatResponseresponse){// 我在中间插了一步：先记账meteringService.recordUsage(response);// 然后转交给原来的 handlerhandler.onCompleteResponse(response);}@OverridepublicvoidonError(Throwableerror){handler.onError(error);}};// 把偷换后的 handler 交给下一层delegate.doChat(request,proxyHandler);}}
- 就是new了一个新接收器，填上一堆自己要的逻辑，然后传给最下层
- 记住调用的时候调的是最上层
现阶段只需知道：正文、思考、工具调用是三条独立的流，不要指望在onPartialResponse里收到思考内容
