序
本文主要研究一下langchain4j的HTTP Client
langchain4j-http-client
langchain4j提供了langchain4j-http-client模块，它实现了一个HttpClient SPI（服务提供者接口），其他模块通过该接口调用LLM提供商的REST API。这意味着底层HTTP客户端可以被自定义，通过实现HttpClient SPI，还可以集成任何其他HTTP客户端。目前，有两个现成的实现：
- langchain4j-http-client-jdk模块中的JdkHttpClient。在例如langchain4j-open-ai等有引用的模块中默认使用
- langchain4j-http-client-spring-restclient模块中的SpringRestClient。在例如langchain4j-open-ai-spring-boot-starter等有引用的spring boot starter中默认使用
HttpClient
langchain4j-http-client/src/main/java/dev/langchain4j/http/client/HttpClient.java
@Experimental
public interface HttpClient {
/**
* Executes a given HTTP request synchronously and returns the response.
* This method blocks until the entire response is received.
*
* @param request the HTTP request to be executed.
* @return a {@link SuccessfulHttpResponse} containing the response data for successful HTTP requests (2XX status codes)
* @throws HttpException if the server returns a client (4XX) or server (5XX) error response
* @throws RuntimeException if an unexpected error occurs during request execution (e.g., network issues, timeouts)
*/
SuccessfulHttpResponse execute(HttpRequest request) throws HttpException, RuntimeException;
/**
* Executes a given HTTP request asynchronously with server-sent events (SSE) handling.
* This method returns immediately while processing continues on a separate thread.
* Events are processed through the provided {@link ServerSentEventListener}.
* <p>
* The execution flow is as follows:
* <ol>
* <li>The request is initiated asynchronously</li>
* <li>Received SSE data is parsed using the {@link DefaultServerSentEventParser}</li>
* <li>Parsed events are delivered to the listener's appropriate methods</li>
* <li>If an error occurs, {@link ServerSentEventListener#onError(Throwable)} is called</li>
* </ol>
* <p>
* If any exception is thrown from the listener's methods, the stream processing
* will be terminated and no further events will be processed.
*
* @param request the HTTP request to be executed.
* @param listener the listener to receive parsed events and error notifications.
*/
default void execute(HttpRequest request, ServerSentEventListener listener) {
execute(request, new DefaultServerSentEventParser(), listener);
}
/**
* Executes a given HTTP request asynchronously with server-sent events (SSE) handling.
* This method returns immediately while processing continues on a separate thread.
* Events are processed through the provided {@link ServerSentEventListener}.
* <p>
* The execution flow is as follows:
* <ol>
* <li>The request is initiated asynchronously</li>
* <li>Received SSE data is parsed using the provided parser</li>
* <li>Parsed events are delivered to the listener's appropriate methods</li>
* <li>If an error occurs, {@link ServerSentEventListener#onError(Throwable)} is called</li>
* </ol>
* <p>
* If any exception is thrown from the listener's methods, the stream processing
* will be terminated and no further events will be processed.
*
* @param request the HTTP request to be executed.
* @param parser the parser to process incoming server-sent events.
* @param listener the listener to receive parsed events and error notifications.
*/
void execute(HttpRequest request, ServerSentEventParser parser, ServerSentEventListener listener);
}
HttpClient定义了execute方法，其中有一个支持ServerSentEventListener
HttpClientBuilderFactory
langchain4j-http-client/src/main/java/dev/langchain4j/http/client/HttpClientBuilderFactory.java
@Experimental
public interface HttpClientBuilderFactory {
HttpClientBuilder create();
}
HttpClientBuilderFactory定义了create方法返回HttpClientBuilder
HttpClientBuilder
langchain4j-http-client/src/main/java/dev/langchain4j/http/client/HttpClientBuilder.java
@Experimental
public interface HttpClientBuilder {
Duration connectTimeout();
HttpClientBuilder connectTimeout(Duration timeout);
Duration readTimeout();
HttpClientBuilder readTimeout(Duration timeout);
HttpClient build();
}
HttpClientBuilder定义了connectTimeout、readTimeout、build方法
HttpClientBuilderLoader
langchain4j-http-client/src/main/java/dev/langchain4j/http/client/HttpClientBuilderLoader.java
@Experimental
public class HttpClientBuilderLoader {
public static HttpClientBuilder loadHttpClientBuilder() {
Collection<HttpClientBuilderFactory> factories = loadFactories(HttpClientBuilderFactory.class);
if (factories.size() > 1) {
List<String> factoryNames = factories.stream()
.map(factory -> factory.getClass().getName())
.toList();
throw new IllegalStateException(String.format("Conflict: multiple HTTP clients have been found " +
"in the classpath: %s. Please explicitly specify the one you wish to use.", factoryNames));
}
for (HttpClientBuilderFactory factory : factories) {
return factory.create();
}
throw new IllegalStateException("No HTTP client has been found in the classpath");
}
}
HttpClientBuilderLoader提供了loadHttpClientBuilder方法，它通过dev.langchain4j.spi.ServiceHelper.loadFactories方法来加载HttpClientBuilderFactory的实现类
ServiceHelper
langchain4j-core/src/main/java/dev/langchain4j/spi/ServiceHelper.java
/**
* Load all the services of a given type.
*
* @param clazz the type of service
* @param <T> the type of service
* @return the list of services, empty if none
*/
public static <T> Collection<T> loadFactories(Class<T> clazz) {
return loadFactories(clazz, null);
}
/**
* Load all the services of a given type.
*
* <p>Utility mechanism around {@code ServiceLoader.load()}</p>
*
* <ul>
* <li>If classloader is {@code null}, will try {@code ServiceLoader.load(clazz)}</li>
* <li>If classloader is not {@code null}, will try {@code ServiceLoader.load(clazz, classloader)}</li>
* </ul>
*
* <p>If the above return nothing, will fall back to {@code ServiceLoader.load(clazz, $this class loader$)}</p>
*
* @param clazz the type of service
* @param classLoader the classloader to use, may be null
* @param <T> the type of service
* @return the list of services, empty if none
*/
public static <T> Collection<T> loadFactories(Class<T> clazz, /* @Nullable */ ClassLoader classLoader) {
List<T> result;
if (classLoader != null) {
result = loadAll(ServiceLoader.load(clazz, classLoader));
} else {
// this is equivalent to:
// ServiceLoader.load(clazz, TCCL);
result = loadAll(ServiceLoader.load(clazz));
}
if (result.isEmpty()) {
// By default, ServiceLoader.load uses the TCCL, this may not be enough in environment dealing with
// classloaders differently such as OSGi. So we should try to use the classloader having loaded this
// class. In OSGi it would be the bundle exposing vert.x and so have access to all its classes.
result = loadAll(ServiceLoader.load(clazz, ServiceHelper.class.getClassLoader()));
}
return result;
}
ServiceHelper是对JDK自带的ServiceLoader.load()进行了封装
langchain4j-http-client-jdk
JdkHttpClientBuilderFactory
dev/langchain4j/http/client/jdk/JdkHttpClientBuilderFactory.java
public class JdkHttpClientBuilderFactory implements HttpClientBuilderFactory {
@Override
public JdkHttpClientBuilder create() {
return JdkHttpClient.builder();
}
}
JdkHttpClientBuilderFactory的create返回的是JdkHttpClientBuilder
JdkHttpClientBuilder
dev/langchain4j/http/client/jdk/JdkHttpClientBuilder.java
public class JdkHttpClientBuilder implements HttpClientBuilder {
private java.net.http.HttpClient.Builder httpClientBuilder;
private Duration connectTimeout;
private Duration readTimeout;
public java.net.http.HttpClient.Builder httpClientBuilder() {
return httpClientBuilder;
}
public JdkHttpClientBuilder httpClientBuilder(java.net.http.HttpClient.Builder httpClientBuilder) {
this.httpClientBuilder = httpClientBuilder;
return this;
}
@Override
public Duration connectTimeout() {
return connectTimeout;
}
@Override
public JdkHttpClientBuilder connectTimeout(Duration connectTimeout) {
this.connectTimeout = connectTimeout;
return this;
}
@Override
public Duration readTimeout() {
return readTimeout;
}
@Override
public JdkHttpClientBuilder readTimeout(Duration readTimeout) {
this.readTimeout = readTimeout;
return this;
}
@Override
public JdkHttpClient build() {
return new JdkHttpClient(this);
}
}
JdkHttpClientBuilder的build构建的是JdkHttpClient
JdkHttpClient
dev/langchain4j/http/client/jdk/JdkHttpClient.java
public class JdkHttpClient implements HttpClient {
private final java.net.http.HttpClient delegate;
private final Duration readTimeout;
public JdkHttpClient(JdkHttpClientBuilder builder) {
java.net.http.HttpClient.Builder httpClientBuilder =
getOrDefault(builder.httpClientBuilder(), java.net.http.HttpClient::newBuilder);
if (builder.connectTimeout() != null) {
httpClientBuilder.connectTimeout(builder.connectTimeout());
}
this.delegate = httpClientBuilder.build();
this.readTimeout = builder.readTimeout();
}
public static JdkHttpClientBuilder builder() {
return new JdkHttpClientBuilder();
}
@Override
public SuccessfulHttpResponse execute(HttpRequest request) throws HttpException {
try {
java.net.http.HttpRequest jdkRequest = toJdkRequest(request);
java.net.http.HttpResponse<String> jdkResponse = delegate.send(jdkRequest, BodyHandlers.ofString());
if (!isSuccessful(jdkResponse)) {
throw new HttpException(jdkResponse.statusCode(), jdkResponse.body());
}
return fromJdkResponse(jdkResponse, jdkResponse.body());
} catch (IOException | InterruptedException e) {
throw new RuntimeException(e);
}
}
@Override
public void execute(HttpRequest request, ServerSentEventParser parser, ServerSentEventListener listener) {
java.net.http.HttpRequest jdkRequest = toJdkRequest(request);
delegate.sendAsync(jdkRequest, BodyHandlers.ofInputStream())
.thenAccept(jdkResponse -> {
if (!isSuccessful(jdkResponse)) {
listener.onError(new HttpException(jdkResponse.statusCode(), readBody(jdkResponse)));
return;
}
SuccessfulHttpResponse response = fromJdkResponse(jdkResponse, null);
listener.onOpen(response);
try (InputStream inputStream = jdkResponse.body()) {
parser.parse(inputStream, listener);
listener.onClose();
} catch (IOException e) {
throw new RuntimeException(e);
}
})
.exceptionally(throwable -> {
if (throwable.getCause() instanceof HttpTimeoutException) {
listener.onError(throwable);
}
return null;
});
}
//......
}
JdkHttpClient的构造器创建了java.net.http.HttpClient作为delegate
langchain4j-http-client-spring-restclient
SpringRestClientBuilderFactory
dev/langchain4j/http/client/spring/restclient/SpringRestClientBuilderFactory.java
public class SpringRestClientBuilderFactory implements HttpClientBuilderFactory {
@Override
public SpringRestClientBuilder create() {
return SpringRestClient.builder();
}
}
SpringRestClientBuilderFactory的create返回的是SpringRestClientBuilder
SpringRestClientBuilder
dev/langchain4j/http/client/spring/restclient/SpringRestClientBuilder.java
public class SpringRestClientBuilder implements HttpClientBuilder {
private RestClient.Builder restClientBuilder;
private AsyncTaskExecutor streamingRequestExecutor;
private Boolean createDefaultStreamingRequestExecutor = true;
private Duration connectTimeout;
private Duration readTimeout;
public RestClient.Builder restClientBuilder() {
return restClientBuilder;
}
public SpringRestClientBuilder restClientBuilder(RestClient.Builder restClientBuilder) {
this.restClientBuilder = restClientBuilder;
return this;
}
public AsyncTaskExecutor streamingRequestExecutor() {
return streamingRequestExecutor;
}
public SpringRestClientBuilder streamingRequestExecutor(AsyncTaskExecutor streamingRequestExecutor) {
this.streamingRequestExecutor = streamingRequestExecutor;
return this;
}
public Boolean createDefaultStreamingRequestExecutor() {
return createDefaultStreamingRequestExecutor;
}
public SpringRestClientBuilder createDefaultStreamingRequestExecutor(Boolean createDefaultStreamingRequestExecutor) {
this.createDefaultStreamingRequestExecutor = createDefaultStreamingRequestExecutor;
return this;
}
@Override
public Duration connectTimeout() {
return connectTimeout;
}
@Override
public SpringRestClientBuilder connectTimeout(Duration connectTimeout) {
this.connectTimeout = connectTimeout;
return this;
}
@Override
public Duration readTimeout() {
return readTimeout;
}
@Override
public SpringRestClientBuilder readTimeout(Duration readTimeout) {
this.readTimeout = readTimeout;
return this;
}
@Override
public SpringRestClient build() {
return new SpringRestClient(this);
}
}
SpringRestClientBuilder的build创建的是SpringRestClient
SpringRestClient
dev/langchain4j/http/client/spring/restclient/SpringRestClient.java
public class SpringRestClient implements HttpClient {
private final RestClient delegate;
private final AsyncTaskExecutor streamingRequestExecutor;
public SpringRestClient(SpringRestClientBuilder builder) {
RestClient.Builder restClientBuilder = getOrDefault(builder.restClientBuilder(), RestClient::builder);
ClientHttpRequestFactorySettings settings = ClientHttpRequestFactorySettings.DEFAULTS;
if (builder.connectTimeout() != null) {
settings = settings.withConnectTimeout(builder.connectTimeout());
}
if (builder.readTimeout() != null) {
settings = settings.withReadTimeout(builder.readTimeout());
}
ClientHttpRequestFactory clientHttpRequestFactory = ClientHttpRequestFactories.get(settings);
this.delegate = restClientBuilder
.requestFactory(clientHttpRequestFactory)
.build();
this.streamingRequestExecutor = getOrDefault(builder.streamingRequestExecutor(), () -> {
if (builder.createDefaultStreamingRequestExecutor()) {
return createDefaultStreamingRequestExecutor();
} else {
return null;
}
});
}
//......
@Override
public SuccessfulHttpResponse execute(HttpRequest request) throws HttpException {
try {
ResponseEntity<String> responseEntity = toSpringRestClientRequest(request)
.retrieve()
.toEntity(String.class);
return SuccessfulHttpResponse.builder()
.statusCode(responseEntity.getStatusCode().value())
.headers(responseEntity.getHeaders())
.body(responseEntity.getBody())
.build();
} catch (RestClientResponseException e) {
throw new HttpException(e.getStatusCode().value(), e.getMessage());
}
}
@Override
public void execute(HttpRequest request, ServerSentEventParser parser, ServerSentEventListener listener) {
streamingRequestExecutor.execute(() -> {
try {
toSpringRestClientRequest(request)
.exchange((springRequest, springResponse) -> {
int statusCode = springResponse.getStatusCode().value();
if (!springResponse.getStatusCode().is2xxSuccessful()) {
String body = springResponse.bodyTo(String.class);
listener.onError(new HttpException(statusCode, body));
return null;
}
SuccessfulHttpResponse response = SuccessfulHttpResponse.builder()
.statusCode(statusCode)
.headers(springResponse.getHeaders())
.build();
listener.onOpen(response);
try (InputStream inputStream = springResponse.getBody()) {
parser.parse(inputStream, listener);
listener.onClose();
}
return null;
});
} catch (Exception e) {
if (e.getCause() instanceof SocketTimeoutException) {
listener.onError(e);
}
}
});
}
}
SpringRestClient的构造器构建了org.springframework.web.client.RestClient为delegate，同时要构建了streamingRequestExecutor用于ServerSentEventListener
小结
langchain4j的langchain4j-http-client模块定义了HttpClient SPI（服务提供者接口），目前有两个现成的实现：
- langchain4j-http-client-jdk模块中的JdkHttpClient，使用的是java.net.http.HttpClient
- langchain4j-http-client-spring-restclient模块中的SpringRestClient，使用的是org.springframework.web.client.RestClient