EasyFlow-Tech-Dev/Easy-Agents
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

发布 v1.1.0 #2

Merged
czm merged 34 commits from develop into main 2026-08-20 11:35:40 +08:00
Conversation 0 Commits 34 Files Changed 266 +373 -3
7 changed files with 373 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 15adcfff42 - Show all commits

134
easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruClient.java
View File

@@ -5,6 +5,7 @@ import com.easyagents.core.util.StringUtil;
import com.easyagents.document.core.exception.DocumentParseException;
import com.easyagents.document.core.entity.ParseFile;
import com.easyagents.document.core.entity.ParseRequest;
import okhttp3.Call;
import okhttp3.MediaType;
import okhttp3.MultipartBody;
import okhttp3.OkHttpClient;
@@ -29,10 +30,17 @@ import java.util.concurrent.TimeUnit;
public class MineruClient {
private static final MediaType DEFAULT_MEDIA_TYPE = MediaType.parse("application/octet-stream");
private static final int DEFAULT_SUBMIT_TIMEOUT_MS = 120000;
private static final int AVAILABILITY_PROBE_TIMEOUT_MS = 3000;
private static final long AVAILABILITY_CACHE_TTL_MS = 5000L;
private final String baseUrl;
private final OkHttpClient okHttpClient;
private final MineruMapper mineruMapper;
private final int submitTimeoutMs;
private final Object availabilityProbeMonitor = new Object();
private volatile long availabilityCacheDeadlineMs;
private volatile String availabilityFailureMessage;
/**
* 创建客户端。
@@ -66,6 +74,7 @@ public class MineruClient {
this.baseUrl = normalizeBaseUrl(properties.getBaseUrl());
this.okHttpClient = okHttpClient;
this.mineruMapper = mineruMapper;
this.submitTimeoutMs = positiveOrDefault(properties.getSubmitTimeoutMs(), DEFAULT_SUBMIT_TIMEOUT_MS);
}
/**
@@ -85,7 +94,10 @@ public class MineruClient {
* @return 原始任务状态
*/
public MineruTaskStatus submit(ParseRequest request) {
return mineruMapper.toTaskStatus(executeJsonMultipart("/tasks", request, buildAsyncFormFields(request)));
assertServiceAvailable();
return mineruMapper.toTaskStatus(
executeJsonMultipart("/tasks", request, buildAsyncFormFields(request), submitTimeoutMs)
);
}
/**
@@ -131,6 +143,22 @@ public class MineruClient {
}
protected JSONObject executeJsonMultipart(String path, ParseRequest request, Map<String, List<String>> fields) {
return executeJsonMultipart(path, request, fields, 0);
}
/**
* 执行带整次调用超时的 Multipart JSON 请求。
*
* @param path 接口路径
* @param request 解析请求
* @param fields 表单字段
* @param callTimeoutMs 整次调用超时时间，单位毫秒；小于等于 0 时沿用客户端阶段超时
* @return JSON 响应
*/
protected JSONObject executeJsonMultipart(String path,
ParseRequest request,
Map<String, List<String>> fields,
long callTimeoutMs) {
MultipartBody.Builder formBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM);
appendFiles(formBuilder, request.getFiles());
appendStringFields(formBuilder, fields);
@@ -138,7 +166,9 @@ public class MineruClient {
.url(baseUrl + path)
.post(formBuilder.build())
.build();
return executeJsonRequest(path, httpRequest);
return callTimeoutMs > 0
? executeJsonRequest(path, httpRequest, callTimeoutMs)
: executeJsonRequest(path, httpRequest);
}
protected JSONObject executeJsonGet(String path) {
@@ -147,7 +177,28 @@ public class MineruClient {
}
protected JSONObject executeJsonRequest(String path, Request request) {
try (Response response = okHttpClient.newCall(request).execute()) {
return executeJsonRequest(path, request, 0);
}
/**
* 执行 JSON 请求，并可限制连接、写入和读取在内的整次调用时长。
*
* @param path 接口路径
* @param request HTTP 请求
* @param callTimeoutMs 整次调用超时时间，单位毫秒；小于等于 0 时不额外限制
* @return JSON 响应
*/
protected JSONObject executeJsonRequest(String path, Request request, long callTimeoutMs) {
Call call = okHttpClient.newCall(request);
if (callTimeoutMs > 0) {
// OkHttp 的 Call timeout 到期后会取消底层请求，避免线程长期阻塞在文件上传或响应等待。
call.timeout().timeout(callTimeoutMs, TimeUnit.MILLISECONDS);
}
try (Response response = call.execute()) {
if (response.code() >= 500) {
// 服务端错误在响应头到达时立即失败，避免异常响应体未结束时继续占用提交线程。
throw buildHttpException(path, response.code(), new byte[0]);
}
ResponseBody body = response.body();
String bodyText = body == null ? "" : body.string();
if (!response.isSuccessful()) {
@@ -163,6 +214,83 @@ public class MineruClient {
}
}
/**
* 在上传文件前探测 MinerU 网关是否可用，避免服务已返回 5xx 时仍传输大文件。
*
* <p>探测结果短暂缓存，限制批量导入期间的额外请求量。健康检查返回 4xx 说明网关仍可达，
* 实际任务接口会继续完成业务校验。</p>
*
* @throws DocumentParseException 网关返回 5xx 或探测请求失败
*/
private void assertServiceAvailable() {
long now = System.currentTimeMillis();
if (now < availabilityCacheDeadlineMs) {
throwCachedAvailabilityFailure();
return;
}
synchronized (availabilityProbeMonitor) {
now = System.currentTimeMillis();
if (now < availabilityCacheDeadlineMs) {
throwCachedAvailabilityFailure();
return;
}
String healthPath = "/health";
Request request = new Request.Builder().url(baseUrl + healthPath).get().build();
Call call = okHttpClient.newCall(request);
call.timeout().timeout(
Math.min(submitTimeoutMs, AVAILABILITY_PROBE_TIMEOUT_MS),
TimeUnit.MILLISECONDS
);
try (Response response = call.execute()) {
if (response.code() >= 500) {
cacheAvailabilityFailure(
"MinerU service unavailable: path=" + healthPath + ", status=" + response.code(),
now
);
throw new DocumentParseException(availabilityFailureMessage);
}
availabilityFailureMessage = null;
availabilityCacheDeadlineMs = now + AVAILABILITY_CACHE_TTL_MS;
} catch (IOException exception) {
cacheAvailabilityFailure("MinerU service unavailable: availability probe failed", now);
throw new DocumentParseException(availabilityFailureMessage, exception);
}
}
}
/**
* 缓存 MinerU 不可用状态。
*
* @param message 失败信息
* @param detectedAtMs 检测时间，单位毫秒
*/
private void cacheAvailabilityFailure(String message, long detectedAtMs) {
availabilityFailureMessage = message;
availabilityCacheDeadlineMs = detectedAtMs + AVAILABILITY_CACHE_TTL_MS;
}
/**
* 若缓存记录为不可用则抛出稳定异常。
*
* @throws DocumentParseException MinerU 仍处于不可用缓存窗口
*/
private void throwCachedAvailabilityFailure() {
if (availabilityFailureMessage != null) {
throw new DocumentParseException(availabilityFailureMessage);
}
}
/**
* 返回正整数配置，非法配置回退到缺省值。
*
* @param value 配置值
* @param defaultValue 缺省值
* @return 可用的正整数
*/
private int positiveOrDefault(Integer value, int defaultValue) {
return value == null || value <= 0 ? defaultValue : value;
}
private void appendFiles(MultipartBody.Builder formBuilder, List<ParseFile> files) {
if (files == null || files.isEmpty()) {
throw new IllegalArgumentException("Parse request must contain at least one file");

19
easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruProperties.java
View File

@@ -16,6 +16,7 @@ public class MineruProperties {
private Integer connectTimeoutMs = 3000;
private Integer readTimeoutMs = 600000;
private Integer writeTimeoutMs = 600000;
private Integer submitTimeoutMs = 120000;
private Integer pollIntervalMs = 1000;
private Integer resultTimeoutMs = 1800000;
private String defaultBackend = "vlm-http-client";
@@ -56,6 +57,24 @@ public class MineruProperties {
this.writeTimeoutMs = writeTimeoutMs;
}
/**
* 获取异步任务提交总超时时间。
*
* @return 提交总超时时间，单位毫秒
*/
public Integer getSubmitTimeoutMs() {
return submitTimeoutMs;
}
/**
* 设置异步任务提交总超时时间。
*
* @param submitTimeoutMs 提交总超时时间，单位毫秒
*/
public void setSubmitTimeoutMs(Integer submitTimeoutMs) {
this.submitTimeoutMs = submitTimeoutMs;
}
public Integer getPollIntervalMs() {
return pollIntervalMs;
}

201
easy-agents-document/easy-agents-document-core/src/test/java/com/easyagents/document/core/mineru/MineruDocumentParseServiceTest.java
View File

@@ -6,14 +6,30 @@ import com.easyagents.document.core.entity.ParseRequest;
import com.easyagents.document.core.entity.ParseResponse;
import com.easyagents.document.core.entity.ParseTaskInfo;
import com.easyagents.document.core.entity.ParseTaskStatus;
import com.easyagents.document.core.exception.DocumentParseException;
import okhttp3.MediaType;
import okhttp3.OkHttpClient;
import okhttp3.Protocol;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.Buffer;
import okio.BufferedSource;
import okio.Okio;
import okio.Source;
import okio.Timeout;
import org.junit.Assert;
import org.junit.Test;
import java.io.ByteArrayOutputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;
@@ -89,6 +105,191 @@ public class MineruDocumentParseServiceTest {
Assert.assertTrue(client.lastMultipartBody.contains("\r\nen\r\n"));
}
/**
* 验证异步任务提交总超时会取消无响应的 HTTP 调用。
*
* @throws Exception 本地测试套接字异常
*/
@Test
public void submitShouldCancelUnresponsiveCallAtConfiguredTimeout() throws Exception {
AtomicReference<Socket> acceptedSocket = new AtomicReference<Socket>();
try (ServerSocket serverSocket = new ServerSocket(0)) {
Thread serverThread = new Thread(() -> {
try {
Socket socket = serverSocket.accept();
acceptedSocket.set(socket);
while (socket.getInputStream().read() >= 0) {
// 持续读取请求但不返回响应，模拟 MinerU 提交接口失去响应。
}
} catch (IOException ignore) {
// 客户端超时取消或测试关闭套接字后结束服务线程。
}
}, "mineru-submit-timeout-test-server");
serverThread.setDaemon(true);
serverThread.start();
MineruProperties properties = defaultProperties();
properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort());
properties.setSubmitTimeoutMs(200);
properties.setConnectTimeoutMs(5000);
properties.setReadTimeoutMs(5000);
properties.setWriteTimeoutMs(5000);
MineruClient client = new MineruClient(properties, new MineruMapper(properties));
long startedAt = System.currentTimeMillis();
try {
client.submit(buildRequest());
Assert.fail("Expected MinerU submit timeout");
} catch (DocumentParseException expected) {
long elapsed = System.currentTimeMillis() - startedAt;
Assert.assertTrue("Submit call should be cancelled promptly, elapsed=" + elapsed, elapsed < 2000);
} finally {
Socket socket = acceptedSocket.get();
if (socket != null) {
socket.close();
}
serverThread.join(1000);
}
}
}
/**
* 验证 MinerU 网关返回 5xx 时会在上传文件前立即失败，并复用短期不可用缓存。
*
* @throws Exception 本地测试套接字异常
*/
@Test
public void submitShouldFailBeforeMultipartUploadWhenProbeReturnsServerError() throws Exception {
AtomicInteger requestCount = new AtomicInteger();
AtomicReference<String> requestLine = new AtomicReference<String>();
Thread serverThread;
try (ServerSocket serverSocket = new ServerSocket(0)) {
serverThread = new Thread(() -> {
while (!serverSocket.isClosed()) {
try (Socket socket = serverSocket.accept();
BufferedReader reader = new BufferedReader(
new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))) {
requestCount.incrementAndGet();
requestLine.set(reader.readLine());
String header;
while ((header = reader.readLine()) != null && !header.isEmpty()) {
// 读取完整请求头后再返回网关错误。
}
byte[] body = "Service Unavailable".getBytes(StandardCharsets.UTF_8);
String responseHeaders = "HTTP/1.1 503 Service Unavailable\r\n"
+ "Content-Type: text/plain\r\n"
+ "Content-Length: " + body.length + "\r\n"
+ "Connection: close\r\n\r\n";
socket.getOutputStream().write(responseHeaders.getBytes(StandardCharsets.UTF_8));
socket.getOutputStream().write(body);
socket.getOutputStream().flush();
} catch (IOException ignore) {
// 测试结束关闭 ServerSocket 后退出服务线程。
}
}
}, "mineru-availability-probe-test-server");
serverThread.setDaemon(true);
serverThread.start();
MineruProperties properties = defaultProperties();
properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort());
MineruClient client = new MineruClient(properties, new MineruMapper(properties));
long startedAt = System.currentTimeMillis();
for (int attempt = 0; attempt < 2; attempt++) {
try {
client.submit(buildRequest());
Assert.fail("Expected MinerU availability failure");
} catch (DocumentParseException expected) {
Assert.assertTrue(expected.getMessage().contains("status=503"));
}
}
long elapsed = System.currentTimeMillis() - startedAt;
Assert.assertTrue("Server error should fail promptly, elapsed=" + elapsed, elapsed < 2000);
}
serverThread.join(1000);
Assert.assertEquals("Cached failure should avoid repeated probes", 1, requestCount.get());
Assert.assertTrue(requestLine.get().startsWith("GET /health HTTP/1.1"));
}
/**
* 验证任务接口返回 5xx 后不会等待或读取异常响应体。
*/
@Test
public void submitShouldThrowServerErrorWithoutReadingResponseBody() {
AtomicInteger requestCount = new AtomicInteger();
AtomicReference<Boolean> errorBodyRead = new AtomicReference<Boolean>(false);
OkHttpClient httpClient = new OkHttpClient.Builder()
.addInterceptor(chain -> {
int currentRequest = requestCount.incrementAndGet();
Response.Builder responseBuilder = new Response.Builder()
.request(chain.request())
.protocol(Protocol.HTTP_1_1);
if (currentRequest == 1) {
return responseBuilder
.code(200)
.message("OK")
.body(ResponseBody.create((MediaType) null, new byte[0]))
.build();
}
ResponseBody trackingBody = new ResponseBody() {
private final BufferedSource source = Okio.buffer(new Source() {
@Override
public long read(Buffer sink, long byteCount) {
errorBodyRead.set(true);
return -1;
}
@Override
public Timeout timeout() {
return Timeout.NONE;
}
@Override
public void close() {
// 无底层资源需要关闭。
}
});
@Override
public MediaType contentType() {
return MediaType.parse("text/plain");
}
@Override
public long contentLength() {
return 19;
}
@Override
public BufferedSource source() {
return source;
}
};
return responseBuilder
.code(500)
.message("Internal Server Error")
.body(trackingBody)
.build();
})
.build();
MineruProperties properties = defaultProperties();
MineruClient client = new MineruClient(properties, httpClient, new MineruMapper(properties));
try {
client.submit(buildRequest());
Assert.fail("Expected MinerU server error");
} catch (DocumentParseException expected) {
Assert.assertTrue(expected.getMessage().contains("status=500"));
}
Assert.assertEquals(2, requestCount.get());
Assert.assertFalse("5xx response body should not be read", errorBodyRead.get());
}
private ParseRequest buildRequest() {
ParseRequest request = new ParseRequest();
request.addFile(ParseFile.of("demo.pptx", "ppt".getBytes(StandardCharsets.UTF_8)));

19
easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/mineru/CommonMineruDocumentProperties.java
View File

@@ -19,6 +19,7 @@ public class CommonMineruDocumentProperties {
private Integer connectTimeoutMs = 3000;
private Integer readTimeoutMs = 600000;
private Integer writeTimeoutMs = 600000;
private Integer submitTimeoutMs = 120000;
private Integer pollIntervalMs = 1000;
private Integer resultTimeoutMs = 1800000;
private String defaultBackend = "vlm-http-client";
@@ -59,6 +60,24 @@ public class CommonMineruDocumentProperties {
this.writeTimeoutMs = writeTimeoutMs;
}
/**
* 获取异步任务提交总超时时间。
*
* @return 提交总超时时间，单位毫秒
*/
public Integer getSubmitTimeoutMs() {
return submitTimeoutMs;
}
/**
* 设置异步任务提交总超时时间。
*
* @param submitTimeoutMs 提交总超时时间，单位毫秒
*/
public void setSubmitTimeoutMs(Integer submitTimeoutMs) {
this.submitTimeoutMs = submitTimeoutMs;
}
public Integer getPollIntervalMs() {
return pollIntervalMs;
}

1
easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pdf/mineru/MineruPdfAutoConfiguration.java
View File

@@ -105,6 +105,7 @@ public class MineruPdfAutoConfiguration {
mineruProperties.setConnectTimeoutMs(useCommon ? commonProperties.getConnectTimeoutMs() : null);
mineruProperties.setReadTimeoutMs(useCommon ? commonProperties.getReadTimeoutMs() : null);
mineruProperties.setWriteTimeoutMs(useCommon ? commonProperties.getWriteTimeoutMs() : null);
mineruProperties.setSubmitTimeoutMs(useCommon ? commonProperties.getSubmitTimeoutMs() : null);
mineruProperties.setPollIntervalMs(useCommon ? commonProperties.getPollIntervalMs() : null);
mineruProperties.setResultTimeoutMs(useCommon ? commonProperties.getResultTimeoutMs() : null);
mineruProperties.setDefaultBackend(useCommon ? commonProperties.getDefaultBackend() : null);

1
easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pptx/MineruPptxAutoConfiguration.java
View File

@@ -53,6 +53,7 @@ public class MineruPptxAutoConfiguration {
mineruProperties.setConnectTimeoutMs(properties.getConnectTimeoutMs());
mineruProperties.setReadTimeoutMs(properties.getReadTimeoutMs());
mineruProperties.setWriteTimeoutMs(properties.getWriteTimeoutMs());
mineruProperties.setSubmitTimeoutMs(properties.getSubmitTimeoutMs());
mineruProperties.setPollIntervalMs(properties.getPollIntervalMs());
mineruProperties.setResultTimeoutMs(properties.getResultTimeoutMs());
mineruProperties.setDefaultBackend(properties.getDefaultBackend());

1
easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/xlsx/MineruXlsxAutoConfiguration.java
View File

@@ -53,6 +53,7 @@ public class MineruXlsxAutoConfiguration {
mineruProperties.setConnectTimeoutMs(properties.getConnectTimeoutMs());
mineruProperties.setReadTimeoutMs(properties.getReadTimeoutMs());
mineruProperties.setWriteTimeoutMs(properties.getWriteTimeoutMs());
mineruProperties.setSubmitTimeoutMs(properties.getSubmitTimeoutMs());
mineruProperties.setPollIntervalMs(properties.getPollIntervalMs());
mineruProperties.setResultTimeoutMs(properties.getResultTimeoutMs());
mineruProperties.setDefaultBackend(properties.getDefaultBackend());