001/* 002 * Copyright (c) 2023-2026, Agents-Flex (fuhai999@gmail.com). 003 * <p> 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * <p> 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * <p> 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 */ 016package com.agentsflex.core.model.chat; 017 018import com.agentsflex.core.model.chat.log.ChatMessageLogger; 019import com.agentsflex.core.model.chat.response.AiMessageResponse; 020import com.agentsflex.core.model.client.ChatClient; 021import com.agentsflex.core.model.client.ChatRequestSpec; 022import com.agentsflex.core.model.client.ChatRequestSpecBuilder; 023import com.agentsflex.core.prompt.Prompt; 024 025import java.util.ArrayList; 026import java.util.Collections; 027import java.util.List; 028 029/** 030 * 支持责任链、统一上下文和协议客户端的聊天模型基类。 031 * <p> 032 * 该类为所有具体的 LLM 实现(如 OpenAI、Qwen、Ollama)提供统一入口,并集成: 033 * <ul> 034 * <li><b>责任链模式</b>:通过 {@link ChatInterceptor} 实现请求拦截、监控、日志等横切逻辑</li> 035 * <li><b>线程上下文管理</b>:通过 {@link ChatContextHolder} 在整个调用链中传递上下文信息</li> 036 * <li><b>协议执行抽象</b>:通过 {@link ChatClient} 解耦协议细节,支持 HTTP/gRPC/WebSocket 等</li> 037 * <li><b>可观测性</b>:自动集成 OpenTelemetry(通过 {@link ChatObservabilityInterceptor})</li> 038 * </ul> 039 * 040 * <h2>架构流程</h2> 041 * <ol> 042 * <li>调用 {@link #chat(Prompt, ChatOptions)} 或 {@link #chatStream(Prompt, StreamResponseListener, ChatOptions)}</li> 043 * <li>构建请求上下文(URL/Headers/Body)并初始化 {@link ChatContext}</li> 044 * <li>构建责任链:可观测性拦截器 → 全局拦截器 → 用户拦截器</li> 045 * <li>责任链执行:每个拦截器可修改 {@link ChatContext},最后由 {@link ChatClient} 执行实际调用</li> 046 * <li>结果返回给调用方</li> 047 * </ol> 048 * 049 * @param <T> 具体的配置类型,必须是 {@link ChatConfig} 的子类 050 */ 051public abstract class BaseChatModel<T extends ChatConfig> implements ChatModel { 052 053 /** 054 * 聊天模型配置,包含 API Key、Endpoint、Model 等信息 055 */ 056 protected final T config; 057 protected ChatClient chatClient; 058 protected ChatRequestSpecBuilder chatRequestSpecBuilder; 059 060 /** 061 * 拦截器链,按执行顺序存储(可观测性 → 全局 → 用户) 062 */ 063 private final List<ChatInterceptor> interceptors; 064 065 /** 066 * 构造一个聊天模型实例,不使用实例级拦截器。 067 * 068 * @param config 聊天模型配置 069 */ 070 public BaseChatModel(T config) { 071 this(config, Collections.emptyList()); 072 } 073 074 /** 075 * 构造一个聊天模型实例,并指定实例级拦截器。 076 * <p> 077 * 实例级拦截器会与全局拦截器(通过 {@link GlobalChatInterceptors} 注册)合并, 078 * 执行顺序为:可观测性拦截器 → 全局拦截器 → 实例拦截器。 079 * 080 * @param config 聊天模型配置 081 * @param userInterceptors 实例级拦截器列表 082 */ 083 public BaseChatModel(T config, List<ChatInterceptor> userInterceptors) { 084 this.config = config; 085 this.interceptors = buildInterceptorChain(userInterceptors); 086 } 087 088 /** 089 * 构建完整的拦截器链。 090 * <p> 091 * 执行顺序: 092 * 1. 可观测性拦截器(最外层,最早执行) 093 * 2. 全局拦截器(通过 GlobalChatInterceptors 注册) 094 * 3. 用户拦截器(实例级) 095 * 096 * @param userInterceptors 用户提供的拦截器列表 097 * @return 按执行顺序排列的拦截器链 098 */ 099 private List<ChatInterceptor> buildInterceptorChain(List<ChatInterceptor> userInterceptors) { 100 List<ChatInterceptor> chain = new ArrayList<>(); 101 102 // 1. 可观测性拦截器(最外层) 103 // 仅在配置启用时添加,负责 OpenTelemetry 追踪和指标上报 104 if (config.isObservabilityEnabled()) { 105 chain.add(new ChatObservabilityInterceptor()); 106 } 107 108 // 2. 全局拦截器(通过 GlobalChatInterceptors 注册) 109 // 适用于所有聊天模型实例的通用逻辑(如全局日志、认证) 110 chain.addAll(GlobalChatInterceptors.getInterceptors()); 111 112 // 3. 用户拦截器(实例级) 113 // 适用于当前实例的特定逻辑 114 if (userInterceptors != null) { 115 chain.addAll(userInterceptors); 116 } 117 118 return chain; 119 } 120 121 /** 122 * 执行同步聊天请求。 123 * <p> 124 * 流程: 125 * 1. 构建请求上下文(URL/Headers/Body) 126 * 2. 初始化线程上下文 {@link ChatContext} 127 * 3. 构建并执行责任链 128 * 4. 返回 LLM 响应 129 * 130 * @param prompt 用户输入的提示 131 * @param options 聊天选项(如流式开关、超时等) 132 * @return LLM 响应结果 133 */ 134 @Override 135 public AiMessageResponse chat(Prompt prompt, ChatOptions options) { 136 if (options == null) { 137 options = new ChatOptions(); 138 } 139 // 强制关闭流式 140 options.setStreaming(false); 141 142 143 ChatRequestSpec request = getChatRequestSpecBuilder().buildRequest(prompt, options, config); 144 145 // 初始化聊天上下文(自动清理) 146 try (ChatContextHolder.ChatContextScope scope = 147 ChatContextHolder.beginChat(prompt, options, request, config)) { 148 // 构建同步责任链并执行 149 SyncChain chain = buildSyncChain(0); 150 return chain.proceed(this, scope.context); 151 } 152 } 153 154 /** 155 * 执行流式聊天请求。 156 * <p> 157 * 流程与同步请求类似,但返回结果通过回调方式分片返回。 158 * 159 * @param prompt 用户输入的提示 160 * @param listener 流式响应监听器 161 * @param options 聊天选项 162 */ 163 @Override 164 public void chatStream(Prompt prompt, StreamResponseListener listener, ChatOptions options) { 165 if (options == null) { 166 options = new ChatOptions(); 167 } 168 options.setStreaming(true); 169 170 ChatRequestSpec request = getChatRequestSpecBuilder().buildRequest(prompt, options, config); 171 172 try (ChatContextHolder.ChatContextScope scope = 173 ChatContextHolder.beginChat(prompt, options, request, config)) { 174 175 StreamChain chain = buildStreamChain(0); 176 chain.proceed(this, scope.context, listener); 177 } 178 } 179 180 181 /** 182 * 构建同步责任链。 183 * <p> 184 * 递归构建拦截器链,链尾节点负责创建并调用 {@link ChatClient}。 185 * 186 * @param index 当前拦截器索引 187 * @return 同步责任链 188 */ 189 private SyncChain buildSyncChain(int index) { 190 // 链尾:执行实际 LLM 调用 191 if (index >= interceptors.size()) { 192 return (model, context) -> { 193 AiMessageResponse aiMessageResponse = null; 194 try { 195 ChatMessageLogger.logRequest(model.getConfig(), context.getRequestSpec().getBody()); 196 aiMessageResponse = getChatClient().chat(); 197 return aiMessageResponse; 198 } finally { 199 ChatMessageLogger.logResponse(model.getConfig(), aiMessageResponse == null ? "" : aiMessageResponse.getRawText()); 200 } 201 }; 202 } 203 204 // 递归构建下一个节点 205 ChatInterceptor current = interceptors.get(index); 206 SyncChain next = buildSyncChain(index + 1); 207 208 // 当前节点:执行拦截器逻辑 209 return (model, context) -> current.intercept(model, context, next); 210 } 211 212 /** 213 * 构建流式责任链。 214 * <p> 215 * 与同步链类似,但支持流式监听器。 216 * 217 * @param index 当前拦截器索引 218 * @return 流式责任链 219 */ 220 private StreamChain buildStreamChain(int index) { 221 if (index >= interceptors.size()) { 222 return (model, context, listener) -> { 223 getChatClient().chatStream(listener); 224 }; 225 } 226 227 ChatInterceptor current = interceptors.get(index); 228 StreamChain next = buildStreamChain(index + 1); 229 return (model, context, listener) -> current.interceptStream(model, context, listener, next); 230 } 231 232 233 public T getConfig() { 234 return config; 235 } 236 237 public ChatClient getChatClient() { 238 return chatClient; 239 } 240 241 public void setChatClient(ChatClient chatClient) { 242 this.chatClient = chatClient; 243 } 244 245 public ChatRequestSpecBuilder getChatRequestSpecBuilder() { 246 return chatRequestSpecBuilder; 247 } 248 249 public void setChatRequestSpecBuilder(ChatRequestSpecBuilder chatRequestSpecBuilder) { 250 this.chatRequestSpecBuilder = chatRequestSpecBuilder; 251 } 252 253 public List<ChatInterceptor> getInterceptors() { 254 return interceptors; 255 } 256 257 /** 258 * 动态添加拦截器。 259 * <p> 260 * 新拦截器会被添加到链的末尾(在用户拦截器区域)。 261 * 262 * @param interceptor 要添加的拦截器 263 */ 264 public void addInterceptor(ChatInterceptor interceptor) { 265 interceptors.add(interceptor); 266 } 267 268 public void addInterceptor(int index, ChatInterceptor interceptor) { 269 interceptors.add(index, interceptor); 270 } 271}