聊聊langchain4j的HTTP Client

序

本文主要研究一下langchain4j的HTTP Client

httpclient_3">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()进行了封装

httpclientjdk_181">langchain4j-http-client-jdk

JdkHttpClientBuilderFactory

dev/langchain4j/http/client/jdk/JdkHttpClientBuilderFactory.java

public class JdkHttpClientBuilderFactory implements HttpClientBuilderFactory {@Overridepublic 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;}@Overridepublic Duration connectTimeout() {return connectTimeout;}@Overridepublic JdkHttpClientBuilder connectTimeout(Duration connectTimeout) {this.connectTimeout = connectTimeout;return this;}@Overridepublic Duration readTimeout() {return readTimeout;}@Overridepublic JdkHttpClientBuilder readTimeout(Duration readTimeout) {this.readTimeout = readTimeout;return this;}@Overridepublic 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();}@Overridepublic 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);}}@Overridepublic 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

httpclientspringrestclient_317">langchain4j-http-client-spring-restclient

SpringRestClientBuilderFactory

dev/langchain4j/http/client/spring/restclient/SpringRestClientBuilderFactory.java

public class SpringRestClientBuilderFactory implements HttpClientBuilderFactory {@Overridepublic 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;}@Overridepublic Duration connectTimeout() {return connectTimeout;}@Overridepublic SpringRestClientBuilder connectTimeout(Duration connectTimeout) {this.connectTimeout = connectTimeout;return this;}@Overridepublic Duration readTimeout() {return readTimeout;}@Overridepublic SpringRestClientBuilder readTimeout(Duration readTimeout) {this.readTimeout = readTimeout;return this;}@Overridepublic 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;}});}//......@Overridepublic 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());}}@Overridepublic 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