序

本文主要研究一下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

doc

• customizable-http-client