Compare commits

...

2 Commits

Author SHA1 Message Date
588e810c51 feat: 接入一库一工具知识检索
- 编译知识库英文运行名、描述、检索配置和独立 Registration

- 统一最终分数阈值并保持模型上下文、检索事件与引用一致

- 完善 AG-UI 知识库检索运行态与完成态展示
2026-08-29 15:45:07 +08:00
30904f1503 fix: 支持可信内部文件引用读取
- 通过服务端文件记录和存储配置恢复可信物理读取句柄

- 文档解析与工作流文档节点优先读取已确认的内部存储对象

- 补齐路径校验、记录异常和读取边界测试
2026-08-29 15:45:02 +08:00
23 changed files with 1348 additions and 56 deletions

View File

@@ -11,6 +11,7 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -151,6 +152,19 @@ public class FileStorageManager implements FileStorageService {
return serviceForHandle(handle).readRecoverable(handle);
}
/**
* 使用当前后端解析服务端可信文件引用。
*
* @param reference 文件 URL 或其他后端可识别引用
* @return 可信文件的物理读取句柄;引用无法确认时为空
* @throws IOException 文件记录无法安全解析时抛出
*/
@Override
public Optional<FileStorageWriteHandle> resolveTrustedFile(String reference)
throws IOException {
return currentService().resolveTrustedFile(reference);
}
/**
* 严格按句柄中的后端精确删除物理对象。
*

View File

@@ -5,6 +5,7 @@ import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Optional;
/**
* EasyFlow 文件存储统一接口。
@@ -105,6 +106,21 @@ public interface FileStorageService {
throw unsupportedRecoverableOperation("readRecoverable");
}
/**
* 将服务端可信文件引用解析为物理读取句柄。
*
* <p>实现必须以服务端持久化记录或存储平台配置为信任来源,并要求外部引用与可信来源
* 精确匹配;不得仅根据客户端传入的 URL、路径或 locator 构造句柄。</p>
*
* @param reference 文件 URL 或其他后端可识别引用
* @return 可信文件的物理读取句柄;引用无法确认时为空
* @throws IOException 文件记录损坏或存储配置不兼容时抛出
*/
default Optional<FileStorageWriteHandle> resolveTrustedFile(String reference)
throws IOException {
return Optional.empty();
}
/**
* 精确且幂等地删除句柄对应的物理对象。
*

View File

@@ -21,6 +21,7 @@ import java.io.*;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Objects;
import java.util.Optional;
/**
* 基于 x-file-storage 的 EasyFlow 文件存储实现。
@@ -268,6 +269,47 @@ public class XFIleStorageServiceImpl implements FileStorageService {
}
}
/**
* 使用 x-file-storage 文件记录或服务端平台配置恢复可信物理读取句柄。
*
* <p>优先使用 recorder 的精确记录。记录不存在时,仅允许与已配置平台 domain、basePath
* 及重建后的完整 URL 完全一致的引用。其他 URL 返回空,由上层继续执行公网地址安全校验。</p>
*
* @param reference 文件 URL
* @return 可信文件的物理读取句柄;引用无法确认时为空
* @throws IOException 文件记录损坏或平台配置不兼容时抛出
*/
@Override
public Optional<FileStorageWriteHandle> resolveTrustedFile(String reference)
throws IOException {
if (!StringUtils.hasText(reference)) {
return Optional.empty();
}
FileInfo fileInfo = null;
try {
fileInfo = fileStorageService.getFileInfoByUrl(reference);
} catch (RuntimeException exception) {
// 存储平台配置本身仍可提供精确可信边界,记录器异常不应阻断内部对象读取。
LOG.warn("查询 x-file-storage 文件记录失败继续按服务端存储配置识别reference={}",
reference, exception);
}
if (fileInfo != null) {
if (!reference.equals(fileInfo.getUrl())) {
throw new IOException("x-file-storage 文件记录 URL 与请求引用不一致");
}
try {
FileStorageWriteHandle handle = handleFromFileInfo(fileInfo);
FileStorage storage = requireStorage(handle);
requirePersistedBasePathSupport(storage, handle);
verifyRecordedLocation(fileInfo, handle);
return Optional.of(handle);
} catch (RuntimeException exception) {
throw new IOException("x-file-storage 文件记录无法恢复为安全读取位置", exception);
}
}
return resolveConfiguredStorageReference(reference);
}
/**
* 直接调用句柄指定平台的物理删除与存在检查,绕过依赖 URL 记录的聚合删除路径。
*
@@ -391,6 +433,108 @@ public class XFIleStorageServiceImpl implements FileStorageService {
}
}
/**
* 校验 recorder 中的物理定位字段可由恢复句柄无损重建。
*
* @param fileInfo 服务端文件记录
* @param handle 恢复出的物理读取句柄
*/
private void verifyRecordedLocation(FileInfo fileInfo, FileStorageWriteHandle handle) {
String actualBasePath = fileInfo.getBasePath() == null ? "" : fileInfo.getBasePath();
String actualPath = fileInfo.getPath() == null ? "" : fileInfo.getPath();
if (!handle.getPlatform().equals(fileInfo.getPlatform())
|| !handle.getBasePath().equals(actualBasePath)
|| !physicalPath(handle).equals(actualPath)
|| !handle.getFilename().equals(fileInfo.getFilename())) {
throw new IllegalStateException("x-file-storage 文件记录包含非规范物理位置");
}
}
/**
* 从 FileInfo 的物理定位字段构造严格校验的读取句柄。
*
* @param fileInfo 服务端文件信息
* @return 可信物理读取句柄
*/
private FileStorageWriteHandle handleFromFileInfo(FileInfo fileInfo) {
String basePath = fileInfo.getBasePath() == null ? "" : fileInfo.getBasePath();
return new FileStorageWriteHandle(
RECOVERABLE_BACKEND,
fileInfo.getPlatform(),
basePath,
recordedRelativePath(basePath, fileInfo.getPath()),
fileInfo.getFilename());
}
/**
* 按服务端配置的平台 domain 与 basePath 识别内部存储 URL。
*
* <p>解析后会再次通过平台自身的 getFileKey 重建完整 URL 并进行精确比较,避免仅凭
* host 或字符串前缀放行其他私网目标。</p>
*
* @param reference 待识别 URL
* @return 精确匹配配置的读取句柄;不匹配任何平台时为空
* @throws IOException 匹配平台前缀但路径无法安全恢复时抛出
*/
private Optional<FileStorageWriteHandle> resolveConfiguredStorageReference(
String reference) throws IOException {
if (fileStorageService.getFileStorageList() == null) {
return Optional.empty();
}
for (FileStorage storage : fileStorageService.getFileStorageList()) {
String domain = readDomainBestEffort(storage);
if (!StringUtils.hasText(domain)) {
continue;
}
String basePath = readRequiredBasePath(storage);
String prefix = domain + basePath;
if (!reference.startsWith(prefix)) {
continue;
}
try {
String remainder = reference.substring(prefix.length());
int filenameIndex = remainder.lastIndexOf('/') + 1;
FileInfo fileInfo = new FileInfo()
.setUrl(reference)
.setPlatform(storage.getPlatform())
.setBasePath(basePath)
.setPath(remainder.substring(0, filenameIndex))
.setFilename(remainder.substring(filenameIndex));
FileStorageWriteHandle handle = handleFromFileInfo(fileInfo);
requirePersistedBasePathSupport(storage, handle);
verifyRecordedLocation(fileInfo, handle);
if (!reference.equals(deriveUrlBestEffort(storage, toFileInfo(handle)))) {
throw new IllegalArgumentException("重建 URL 与请求引用不一致");
}
return Optional.of(handle);
} catch (RuntimeException exception) {
throw new IOException("服务端存储 URL 无法恢复为安全读取位置", exception);
}
}
return Optional.empty();
}
/**
* 将 recorder 保存的 x-file-storage 物理路径还原为句柄相对路径。
*
* @param basePath 平台基础路径
* @param recordedPath recorder 中保存的物理目录
* @return 不带前导斜杠的相对目录
*/
private String recordedRelativePath(String basePath, String recordedPath) {
String path = recordedPath == null ? "" : recordedPath;
if (basePath.isEmpty() || basePath.endsWith("/")) {
if (path.startsWith("/")) {
throw new IllegalArgumentException("文件记录路径与平台基础路径格式不一致");
}
return path;
}
if (!path.startsWith("/")) {
throw new IllegalArgumentException("文件记录路径缺少必要的前导斜杠");
}
return path.substring(1);
}
/**
* 构造仅包含精确物理定位字段的 FileInfo。
*
@@ -487,16 +631,26 @@ public class XFIleStorageServiceImpl implements FileStorageService {
* @return 可推导 URL平台不支持时返回 null
*/
private String deriveUrlBestEffort(FileStorage storage, FileInfo fileInfo) {
String domain = readDomainBestEffort(storage);
if (domain == null) {
return null;
}
return domain + storage.getFileKey(fileInfo);
}
/**
* 使用平台公开的 getDomain 方法读取文件访问域名。
*
* @param storage 具体平台存储
* @return 平台访问域名;平台不支持时返回 null
*/
private String readDomainBestEffort(FileStorage storage) {
try {
Method method = storage.getClass().getMethod("getDomain");
if (!String.class.equals(method.getReturnType())) {
return null;
}
String domain = (String) method.invoke(storage);
if (domain == null) {
return null;
}
return domain + storage.getFileKey(fileInfo);
return (String) method.invoke(storage);
} catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException | RuntimeException exception) {
LOG.debug("当前 x-file-storage 平台无法推导 recorder URL: {}", storage.getClass().getName());
return null;

View File

@@ -7,6 +7,7 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
@@ -49,6 +50,27 @@ public class FileStorageManagerTest {
assertFalse(exists);
}
/**
* 验证服务端文件记录解析使用当前配置的具体存储后端。
*
* @throws IOException 文件记录解析失败时抛出
*/
@Test
public void recordedFileResolutionUsesCurrentBackend() throws IOException {
RecordingStorage local = new RecordingStorage("local");
RecordingStorage xFile = new RecordingStorage("xFileStorage");
FileStorageManager manager = new FileStorageManager(
() -> "xFileStorage",
backend -> Map.of("local", local, "xFileStorage", xFile).get(backend));
Optional<FileStorageWriteHandle> resolved = manager.resolveTrustedFile(
"http://127.0.0.1:39000/easyflow/attachment/demo.pdf");
assertSame(xFile.recordedHandle, resolved.orElseThrow());
assertEquals(1, xFile.resolveCalls);
assertEquals(0, local.resolveCalls);
}
/**
* 可记录可恢复调用的存储测试替身。
*/
@@ -57,6 +79,8 @@ public class FileStorageManagerTest {
private final String backend;
/** 固定结果。 */
private final FileStorageWriteResult result;
/** 固定服务端文件记录句柄。 */
private final FileStorageWriteHandle recordedHandle;
/** 固定可恢复读取流。 */
private final InputStream recoverableInput = InputStream.nullInputStream();
/** prepare 调用次数。 */
@@ -69,6 +93,8 @@ public class FileStorageManagerTest {
private int deleteCalls;
/** exists 调用次数。 */
private int existsCalls;
/** 服务端文件记录解析调用次数。 */
private int resolveCalls;
/**
* 创建指定名称的存储替身。
@@ -80,6 +106,8 @@ public class FileStorageManagerTest {
FileStorageWriteHandle handle = new FileStorageWriteHandle(
backend, "", "/tmp/easyflow", "skill-content", "content.bin");
this.result = new FileStorageWriteResult("/files/content.bin", handle.encodeLocator());
this.recordedHandle = new FileStorageWriteHandle(
backend, "", "/tmp/easyflow", "attachment", "demo.pdf");
}
/** {@inheritDoc} */
@@ -114,6 +142,13 @@ public class FileStorageManagerTest {
return recoverableInput;
}
/** {@inheritDoc} */
@Override
public Optional<FileStorageWriteHandle> resolveTrustedFile(String reference) {
resolveCalls++;
return Optional.of(recordedHandle);
}
/** {@inheritDoc} */
@Override
public void deleteRecoverable(FileStorageWriteHandle handle) {

View File

@@ -212,6 +212,123 @@ public class XFIleStorageServiceImplTest {
client.lastArgs.object());
}
/**
* 验证 recorder 登记的回环地址附件可恢复为可信句柄并通过 MinIO 客户端直读。
*
* @throws Exception 测试替身配置或流读取失败
*/
@Test
public void recordedLoopbackUrlUsesExactMinioObject() throws Exception {
byte[] content = "workflow-content".getBytes(java.nio.charset.StandardCharsets.UTF_8);
RecordingMinioClient client = new RecordingMinioClient(content);
MinioFileStorage platform = new MinioFileStorage();
platform.setPlatform("minio-main");
platform.setBucketName("easyflow");
platform.setBasePath("attachment");
platform.setDomain("http://127.0.0.1:39000/easyflow/");
platform.setClientFactory(new FixedMinioClientFactory(client));
RecoverableStorageService delegate = new RecoverableStorageService(platform);
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/1/2026/8/26/"
+ "d6186b17-4ab7-4f99-9299-b19df7ff0a3b/投标文件否决(废标)违规事项汇总.pdf";
delegate.recordedFileInfo = new FileInfo()
.setUrl(fileUrl)
.setPlatform("minio-main")
.setBasePath("attachment")
.setPath("/1/2026/8/26/d6186b17-4ab7-4f99-9299-b19df7ff0a3b/")
.setFilename("投标文件否决(废标)违规事项汇总.pdf");
XFIleStorageServiceImpl service = createService(delegate);
FileStorageWriteHandle handle = service.resolveTrustedFile(fileUrl).orElseThrow();
byte[] actual;
try (InputStream inputStream = service.readRecoverable(handle)) {
actual = inputStream.readAllBytes();
}
assertEquals("attachment", handle.getBasePath());
assertEquals(
"1/2026/8/26/d6186b17-4ab7-4f99-9299-b19df7ff0a3b/",
handle.getPath());
assertArrayEquals(content, actual);
assertEquals("easyflow", client.lastArgs.bucket());
assertEquals(
"attachment/1/2026/8/26/d6186b17-4ab7-4f99-9299-b19df7ff0a3b/"
+ "投标文件否决(废标)违规事项汇总.pdf",
client.lastArgs.object());
}
/**
* 验证 recorder 没有记录时,服务端配置的存储 URL 仍可通过 MinIO 客户端直读。
*
* @throws Exception 测试替身配置或流读取失败
*/
@Test
public void configuredStorageUrlWithoutRecorderUsesExactMinioObject() throws Exception {
byte[] content = "configured-content".getBytes(java.nio.charset.StandardCharsets.UTF_8);
RecordingMinioClient client = new RecordingMinioClient(content);
MinioFileStorage platform = new MinioFileStorage();
platform.setPlatform("minio-main");
platform.setBucketName("easyflow");
platform.setBasePath("attachment");
platform.setDomain("http://127.0.0.1:39000/easyflow/");
platform.setClientFactory(new FixedMinioClientFactory(client));
XFIleStorageServiceImpl service = createService(new RecoverableStorageService(platform));
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/1/2026/8/26/"
+ "0f0db465-7fa0-46b1-9fae-4f5a8c85f881/投标文件否决(废标)违规事项汇总.pdf";
FileStorageWriteHandle handle = service.resolveTrustedFile(fileUrl).orElseThrow();
byte[] actual;
try (InputStream inputStream = service.readRecoverable(handle)) {
actual = inputStream.readAllBytes();
}
assertArrayEquals(content, actual);
assertEquals(
"attachment/1/2026/8/26/0f0db465-7fa0-46b1-9fae-4f5a8c85f881/"
+ "投标文件否决(废标)违规事项汇总.pdf",
client.lastArgs.object());
}
/**
* 验证未配置为存储地址的回环 URL 不会被识别为可信附件。
*
* @throws Exception 测试替身注入失败
*/
@Test
public void unconfiguredLoopbackUrlIsNotResolved() throws Exception {
RecoverablePlatform platform = new RecoverablePlatform(
"minio-main", "attachment", "http://127.0.0.1:39000/easyflow/");
XFIleStorageServiceImpl service = createService(new RecoverableStorageService(platform));
assertTrue(service.resolveTrustedFile(
"http://127.0.0.1:39000/other/unconfigured.pdf").isEmpty());
}
/**
* 验证 recorder 中非规范物理路径会失败关闭。
*
* @throws Exception 测试替身注入失败
*/
@Test
public void corruptedRecordedLocationIsRejected() throws Exception {
RecoverablePlatform platform = new RecoverablePlatform(
"minio-main", "attachment", "http://127.0.0.1:39000/easyflow/");
RecoverableStorageService delegate = new RecoverableStorageService(platform);
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/demo.pdf";
delegate.recordedFileInfo = new FileInfo()
.setUrl(fileUrl)
.setPlatform("minio-main")
.setBasePath("attachment")
.setPath("missing-leading-slash/")
.setFilename("demo.pdf");
XFIleStorageServiceImpl service = createService(delegate);
IOException exception = assertThrows(
IOException.class,
() -> service.resolveTrustedFile(fileUrl));
assertTrue(exception.getMessage().contains("无法恢复"));
}
/**
* 验证 recorder 完全缺失目标记录时,精确删除仍直接作用于物理平台并成功。
*
@@ -471,6 +588,8 @@ public class XFIleStorageServiceImplTest {
private int recorderDeleteCalls;
/** recorder 删除是否抛出异常。 */
private boolean recorderDeleteThrows;
/** recorder 返回的服务端文件记录。 */
private FileInfo recordedFileInfo;
/**
* 创建聚合服务替身。
@@ -479,10 +598,17 @@ public class XFIleStorageServiceImplTest {
*/
private RecoverableStorageService(FileStorage platform) {
this.platform = platform;
setFileStorageList(new java.util.concurrent.CopyOnWriteArrayList<>(
java.util.List.of(platform)));
setFileRecorder(new FileRecorder() {
@Override public boolean save(FileInfo fileInfo) { return true; }
@Override public void update(FileInfo fileInfo) { }
@Override public FileInfo getByUrl(String url) { return null; }
@Override public FileInfo getByUrl(String url) {
return recordedFileInfo != null
&& url.equals(recordedFileInfo.getUrl())
? recordedFileInfo
: null;
}
@Override public boolean delete(String url) {
recorderDeleteCalls++;
if (recorderDeleteThrows) {
@@ -506,6 +632,15 @@ public class XFIleStorageServiceImplTest {
return platform.getPlatform().equals(name) ? (T) platform : null;
}
/** {@inheritDoc} */
@Override
public FileInfo getFileInfoByUrl(String url) {
return recordedFileInfo != null
&& url.equals(recordedFileInfo.getUrl())
? recordedFileInfo
: null;
}
/** {@inheritDoc} */
@Override
public org.dromara.x.file.storage.core.upload.UploadPretreatment of(Object file) {

View File

@@ -1169,6 +1169,8 @@ public class AgentRunService {
StringBuilder answer = new StringBuilder();
ChatAssistantAccumulator assistantAccumulator = new ChatAssistantAccumulator();
LegacyThinkingTagParser legacyThinkingTagParser = new LegacyThinkingTagParser();
KnowledgeRetrievalStatusTracker knowledgeRetrievalStatusTracker =
new KnowledgeRetrievalStatusTracker();
// 注册 emit 服务
registerEmitterCancellation(requestId, runOutput, chatContext, answer,
assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog);
@@ -1177,7 +1179,7 @@ public class AgentRunService {
if (isAguiCancellationRequested(runOutput)) {
handleRuntimeEvent(cancellationEvent("用户已停止生成"), requestId, runOutput,
answer, assistantAccumulator, legacyThinkingTagParser,
chatContext, finished, persistChatlog);
knowledgeRetrievalStatusTracker, chatContext, finished, persistChatlog);
if (lockHandle != null) {
releaseRunLockQuietly(lockHandle, requestId);
}
@@ -1195,7 +1197,7 @@ public class AgentRunService {
request.setAgentDefinition(bundle.getDefinition());
request.setRuntimeContext(runtimeContext);
request.setToolInvokers(bundle.getToolInvokers());
request.setKnowledgeRetrievers(bundle.getKnowledgeRetrievers());
request.setKnowledgeRegistrations(bundle.getKnowledgeRegistrations());
request.setSessionStore(runtimeSessionStore);
request.setMediaResolver(agentMediaService.runtimeResolver(account));
request.getMetadata().put("assistantCode", assistantCode);
@@ -1224,6 +1226,7 @@ public class AgentRunService {
runRuntimeCallbackSafely(
() -> handleRuntimeEvent(event, requestId, runOutput, answer,
assistantAccumulator, legacyThinkingTagParser,
knowledgeRetrievalStatusTracker,
chatContext, finished, persistChatlog),
requestId, runOutput, chatContext, finished, persistChatlog);
}
@@ -1513,7 +1516,8 @@ public class AgentRunService {
AtomicBoolean finished,
boolean persistChatlog) {
handleRuntimeEvent(event, requestId, runOutput, answer, assistantAccumulator,
new LegacyThinkingTagParser(), chatContext, finished, persistChatlog);
new LegacyThinkingTagParser(), new KnowledgeRetrievalStatusTracker(),
chatContext, finished, persistChatlog);
}
private void handleRuntimeEvent(AgentRuntimeEvent event,
@@ -1525,6 +1529,35 @@ public class AgentRunService {
ChatRuntimeContext chatContext,
AtomicBoolean finished,
boolean persistChatlog) {
handleRuntimeEvent(event, requestId, runOutput, answer, assistantAccumulator,
legacyThinkingTagParser, new KnowledgeRetrievalStatusTracker(),
chatContext, finished, persistChatlog);
}
/**
* 将单个 Runtime 事件投影到聊天协议,并复用本轮知识库工具状态追踪器。
*
* @param event Runtime 事件
* @param requestId 请求 ID
* @param runOutput 运行输出
* @param answer 回答累积器
* @param assistantAccumulator Assistant 结构化累积器
* @param legacyThinkingTagParser 旧思考标签解析器
* @param knowledgeRetrievalStatusTracker 知识库工具状态追踪器
* @param chatContext 聊天上下文
* @param finished 终态仲裁标记
* @param persistChatlog 是否持久化聊天日志
*/
private void handleRuntimeEvent(AgentRuntimeEvent event,
String requestId,
AgentRunOutput runOutput,
StringBuilder answer,
ChatAssistantAccumulator assistantAccumulator,
LegacyThinkingTagParser legacyThinkingTagParser,
KnowledgeRetrievalStatusTracker knowledgeRetrievalStatusTracker,
ChatRuntimeContext chatContext,
AtomicBoolean finished,
boolean persistChatlog) {
if (event == null || event.getEventType() == null) {
return;
}
@@ -1642,6 +1675,17 @@ public class AgentRunService {
return;
}
Map<String, Object> toolPayload = toolStatus;
if (isKnowledgeToolEvent(event)) {
Map<String, Object> statusPayload = buildKnowledgeRetrievalStatusPayload(
knowledgeRetrievalStatusTracker.update(event));
LOG.info("Agent runtime knowledge tool call, requestId={}, toolCallId={}, toolName={}",
requestId, event.getToolCallId(), stringValue(toolPayload, "toolName"));
if (!sendEnvelope(runOutput, ChatDomain.BUSINESS, ChatType.STATUS, statusPayload)) {
cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator,
legacyThinkingTagParser, finished, persistChatlog);
}
return;
}
if (!runOutput.emitRuntimeEvent(publicRuntimeEvent(event, toolPayload))) {
cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator,
legacyThinkingTagParser, finished, persistChatlog);
@@ -1664,6 +1708,20 @@ public class AgentRunService {
}
if (event.getEventType() == AgentRuntimeEventType.TOOL_RESULT) {
Map<String, Object> toolPayload = toolStatus;
if (isKnowledgeToolEvent(event)) {
Map<String, Object> statusPayload = buildKnowledgeRetrievalStatusPayload(
knowledgeRetrievalStatusTracker.update(event));
LOG.info("Agent runtime knowledge tool result, requestId={}, toolCallId={}, toolName={}, status={}",
requestId, event.getToolCallId(), stringValue(toolPayload, "toolName"),
stringValue(statusPayload, "status"));
if (!sendEnvelope(runOutput, ChatDomain.BUSINESS, ChatType.STATUS, statusPayload)) {
cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator,
legacyThinkingTagParser, finished, persistChatlog);
return;
}
legacyThinkingTagParser.reset();
return;
}
LOG.info("Agent runtime tool result, requestId={}, toolCallId={}, toolName={}, status={}",
requestId, event.getToolCallId(), stringValue(toolPayload, "toolName"),
stringValue(toolPayload, "status"));
@@ -1689,10 +1747,7 @@ public class AgentRunService {
if (event.getEventType() == AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL) {
LOG.info("Agent runtime knowledge retrieval, requestId={}, payload={}, metadata={}",
requestId, event.getPayload(), event.getMetadata());
if (!sendEnvelope(runOutput, ChatDomain.BUSINESS, ChatType.STATUS, buildKnowledgeRetrievalStatusPayload(event))) {
cancelDisconnectedRun(requestId, chatContext, answer, assistantAccumulator,
legacyThinkingTagParser, finished, persistChatlog);
}
// 文档摘要事件用于引用与监察UI 完成态统一以 TOOL_RESULT 为准。
return;
}
if (event.getEventType() == AgentRuntimeEventType.MEMORY_COMPRESSION_STARTED
@@ -1750,6 +1805,10 @@ public class AgentRunService {
if (event.getEventType() == AgentRuntimeEventType.FAILED) {
emitAssistantSegments(legacyThinkingTagParser.finish(), requestId, runOutput, chatContext,
answer, assistantAccumulator, legacyThinkingTagParser, finished, persistChatlog);
if (knowledgeRetrievalStatusTracker.failActiveCalls()) {
sendEnvelope(runOutput, ChatDomain.BUSINESS, ChatType.STATUS,
buildKnowledgeRetrievalStatusPayload("error"));
}
runOutput.emitRuntimeEvent(event);
assistantAccumulator.finalizePendingSkillInvocations("FAILED", "技能调用失败");
if (persistChatlog) {
@@ -2813,7 +2872,8 @@ public class AgentRunService {
Map<String, Object> rawPayload = event.getPayload() == null ? Map.of() : event.getPayload();
Map<String, Object> payload = selectPayload(rawPayload,
"name", "status", "success", "toolDisplayName", "toolName",
"skillDisplayName", "skillId");
"skillDisplayName", "skillId", "toolCategory",
"knowledgeId", "knowledgeName", "knowledgeRuntimeName");
String toolCallId = firstText(event.getToolCallId(), stringValue(rawPayload, "toolCallId"));
if (toolCallId != null && !toolCallId.isBlank()) {
payload.put("toolCallId", toolCallId);
@@ -2949,17 +3009,110 @@ public class AgentRunService {
/**
* 构建知识库检索状态载荷,确保前端可按稳定 key 合并同一轮状态行。
*
* @param event 知识库检索运行时事件
* @param status running、done 或 error
* @return 知识库检索状态载荷
*/
private Map<String, Object> buildKnowledgeRetrievalStatusPayload(AgentRuntimeEvent event) {
private Map<String, Object> buildKnowledgeRetrievalStatusPayload(String status) {
String normalizedStatus = "running".equals(status) || "error".equals(status)
? status : "done";
Map<String, Object> payload = new LinkedHashMap<>();
payload.put("statusKey", "knowledge-retrieval");
payload.put("status", "done");
payload.put("label", "已检索知识库");
payload.put("status", normalizedStatus);
payload.put("label", switch (normalizedStatus) {
case "running" -> "正在检索知识库";
case "error" -> "知识库检索失败";
default -> "已检索知识库";
});
return payload;
}
/**
* 判断标准工具生命周期事件是否属于知识库工具。
*
* @param event 运行时工具事件
* @return 知识库工具事件时为 true
*/
private boolean isKnowledgeToolEvent(AgentRuntimeEvent event) {
String category = stringPayload(event, "toolCategory");
if ("KNOWLEDGE".equalsIgnoreCase(category)) {
return true;
}
String toolName = firstText(stringPayload(event, "toolName"), stringPayload(event, "name"));
if (toolName == null) {
return false;
}
String normalizedName = toolName.trim().toLowerCase(Locale.ROOT);
return "retrieve_knowledge".equals(normalizedName)
|| normalizedName.startsWith("retrieve_knowledge_");
}
/**
* 聚合同一批知识库工具调用,避免并行检索中首个结果提前结束 UI 状态。
*/
static final class KnowledgeRetrievalStatusTracker {
private final Set<String> activeToolCallIds = new LinkedHashSet<>();
private boolean failed;
/**
* 应用一次知识库工具生命周期事件。
*
* @param event TOOL_CALL 或 TOOL_RESULT 事件
* @return 聚合后的 running、done 或 error 状态
*/
String update(AgentRuntimeEvent event) {
String toolCallId = toolCallIdentity(event);
if (event.getEventType() == AgentRuntimeEventType.TOOL_CALL) {
if (activeToolCallIds.isEmpty()) {
failed = false;
}
activeToolCallIds.add(toolCallId);
return "running";
}
if (event.getEventType() == AgentRuntimeEventType.TOOL_RESULT) {
activeToolCallIds.remove(toolCallId);
failed = failed || !toolSucceeded(event);
if (!activeToolCallIds.isEmpty()) {
return "running";
}
return failed ? "error" : "done";
}
throw new IllegalArgumentException("Knowledge status only accepts TOOL_CALL or TOOL_RESULT events.");
}
/**
* 将运行失败时仍未结束的知识库调用收口为失败。
*
* @return 存在未结束调用时为 true
*/
boolean failActiveCalls() {
if (activeToolCallIds.isEmpty()) {
return false;
}
activeToolCallIds.clear();
failed = true;
return true;
}
private String toolCallIdentity(AgentRuntimeEvent event) {
String toolCallId = event.getToolCallId();
if (toolCallId == null || toolCallId.isBlank()) {
Object payloadId = event.getPayload() == null ? null : event.getPayload().get("toolCallId");
toolCallId = payloadId == null ? event.getEventId() : String.valueOf(payloadId);
}
return toolCallId;
}
private boolean toolSucceeded(AgentRuntimeEvent event) {
Map<String, Object> payload = event.getPayload() == null ? Map.of() : event.getPayload();
if (Boolean.FALSE.equals(payload.get("success"))) {
return false;
}
Object status = payload.get("status");
return status == null || !"FAILED".equalsIgnoreCase(String.valueOf(status));
}
}
/**
* 构建不含 Runtime 原始上下文的内存压缩公开状态载荷。
*

View File

@@ -1,10 +1,12 @@
package tech.easyflow.agent.runtime;
import com.easyagents.agent.runtime.AgentDefinition;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetriever;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRegistration;
import com.easyagents.agent.runtime.tool.AgentToolInvoker;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
/**
@@ -14,7 +16,7 @@ public class AgentRuntimeBundle {
private AgentDefinition definition;
private Map<String, AgentToolInvoker> toolInvokers = new LinkedHashMap<>();
private Map<String, AgentKnowledgeRetriever> knowledgeRetrievers = new LinkedHashMap<>();
private List<AgentKnowledgeRegistration> knowledgeRegistrations = new ArrayList<>();
/**
* 获取 Agent 定义。
@@ -57,16 +59,18 @@ public class AgentRuntimeBundle {
*
* @return 知识库检索器
*/
public Map<String, AgentKnowledgeRetriever> getKnowledgeRetrievers() {
return knowledgeRetrievers;
public List<AgentKnowledgeRegistration> getKnowledgeRegistrations() {
return knowledgeRegistrations;
}
/**
* 设置知识库检索器。
*
* @param knowledgeRetrievers 知识库检索器
* @param knowledgeRegistrations 知识库运行时绑定
*/
public void setKnowledgeRetrievers(Map<String, AgentKnowledgeRetriever> knowledgeRetrievers) {
this.knowledgeRetrievers = knowledgeRetrievers == null ? new LinkedHashMap<>() : knowledgeRetrievers;
public void setKnowledgeRegistrations(List<AgentKnowledgeRegistration> knowledgeRegistrations) {
this.knowledgeRegistrations = knowledgeRegistrations == null
? new ArrayList<>()
: new ArrayList<>(knowledgeRegistrations);
}
}

View File

@@ -6,9 +6,10 @@ import com.easyagents.agent.runtime.AgentRuntimeContext;
import com.easyagents.agent.runtime.event.AgentRuntimeEvent;
import com.easyagents.agent.runtime.event.AgentRuntimeEventType;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeDocument;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgePolicy;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRegistration;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetrievalResult;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeSpec;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeToolNames;
import com.easyagents.agent.runtime.memory.AgentMemoryCompressionParameter;
import com.easyagents.agent.runtime.memory.AgentMemoryPolicy;
import com.easyagents.agent.runtime.memory.AgentMemoryType;
@@ -118,11 +119,11 @@ public class AgentRuntimeCompiler {
bundle.setDefinition(definition);
compileTools(agent, definition, bundle);
compileKnowledge(agent, definition, bundle);
if (agentBuiltinToolsConfigResolver != null) {
validateBuiltinTools(definition,
agentBuiltinToolsConfigResolver.resolvePublishedRuntime(agent.getExecutionConfigJson()));
}
compileKnowledge(agent, definition, bundle);
return bundle;
}
@@ -294,7 +295,7 @@ public class AgentRuntimeCompiler {
if (config.artifactPublish().enabled()) {
specs.add(buildArtifactPublishSpec(config.artifactPublish()));
}
assertToolBudget(specs, definition.getMcpSpecs());
assertToolBudget(specs, definition.getMcpSpecs(), definition.getKnowledgeSpecs().size());
}
private void attachBuiltinTools(Agent agent,
@@ -510,11 +511,21 @@ public class AgentRuntimeCompiler {
return names;
}
/**
* 校验内置工具与普通工具、知识库工具及 MCP 工具不存在运行名冲突。
*
* @param definition 已编译 Agent 定义
* @param builtinNames 待启用内置工具名称
* @throws BusinessException 工具名称冲突时抛出
*/
private void assertNoBuiltinNameConflict(AgentDefinition definition, Set<String> builtinNames) {
Set<String> existing = new LinkedHashSet<>();
for (AgentToolSpec spec : definition.getToolSpecs()) {
existing.add(spec.getName());
}
for (AgentKnowledgeSpec spec : definition.getKnowledgeSpecs()) {
existing.add(AgentKnowledgeToolNames.build(spec.getRuntimeName()));
}
for (McpSpec mcp : definition.getMcpSpecs()) {
if (mcp.getFrozenToolManifest() != null) {
mcp.getFrozenToolManifest().forEach(entry -> existing.add(entry.getName()));
@@ -540,6 +551,14 @@ public class AgentRuntimeCompiler {
assertToolBudget(toolSpecs, mcpSpecs, 0);
}
/**
* 校验最终工具数量和 Schema 大小预算。
*
* @param toolSpecs 静态 Tool 声明
* @param mcpSpecs MCP 声明
* @param additionalToolCount 知识库等额外工具数量
* @throws BusinessException 超出预算时抛出
*/
private void assertToolBudget(List<AgentToolSpec> toolSpecs,
List<McpSpec> mcpSpecs,
int additionalToolCount) {
@@ -570,7 +589,7 @@ public class AgentRuntimeCompiler {
}
}
if (toolCount > MAX_RUNTIME_TOOL_COUNT) {
throw new BusinessException("Agent Runtime Tool 数量超过 128 个,请减少直接工具或 Skill 绑定");
throw new BusinessException("Agent Runtime Tool 数量超过 128 个,请减少工具、知识库或 Skill 绑定");
}
if (schemaBytes > MAX_RUNTIME_SCHEMA_BYTES) {
throw new BusinessException("Agent Runtime Tool Schema 超过 2 MiB请减少工具或精简 Schema");
@@ -591,12 +610,27 @@ public class AgentRuntimeCompiler {
}
}
/**
* 将 EasyFlow 知识库绑定编译为一库一工具所需的声明和 Retriever 绑定。
*
* @param agent Agent 发布视图
* @param definition 中立 Agent 定义
* @param bundle 运行时编译结果
* @throws BusinessException 知识库不存在、英文运行名非法或工具名冲突时抛出
*/
private void compileKnowledge(Agent agent, AgentDefinition definition, AgentRuntimeBundle bundle) {
if (agent.getKnowledgeBindings() == null) {
return;
}
List<AgentKnowledgeSpec> specs = new ArrayList<>();
Map<String, com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetriever> retrievers = new LinkedHashMap<>();
List<AgentKnowledgeRegistration> registrations = new ArrayList<>();
Set<String> knowledgeToolNames = new LinkedHashSet<>();
Set<String> existingToolNames = new LinkedHashSet<>();
definition.getToolSpecs().stream()
.filter(Objects::nonNull)
.map(AgentToolSpec::getName)
.filter(Objects::nonNull)
.forEach(existingToolNames::add);
for (AgentKnowledgeBinding binding : agent.getKnowledgeBindings()) {
if (!Boolean.TRUE.equals(binding.getEnabled())) {
continue;
@@ -607,9 +641,9 @@ public class AgentRuntimeCompiler {
}
AgentKnowledgeSpec spec = new AgentKnowledgeSpec();
spec.setKnowledgeId(binding.getKnowledgeId().toString());
spec.setRuntimeName(requireKnowledgeRuntimeName(knowledge));
spec.setName(knowledge.getTitle());
spec.setDescription(knowledge.getDescription());
spec.setRetrievalMode(AgentKnowledgePolicy.AGENTIC);
spec.getMetadata().put("knowledgeType", knowledge.getCollectionType());
spec.getMetadata().put("faqCollection", knowledge.isFaqCollection());
Integer limit = intValue(binding.getOptionsJson(), "limit");
@@ -618,11 +652,37 @@ public class AgentRuntimeCompiler {
if (threshold != null) {
spec.setScoreThreshold(threshold);
}
String toolName = AgentKnowledgeToolNames.build(spec.getRuntimeName());
if (!knowledgeToolNames.add(toolName) || existingToolNames.contains(toolName)) {
throw new BusinessException("Agent 知识库工具运行名冲突:" + toolName);
}
specs.add(spec);
retrievers.put(spec.getKnowledgeId(), request -> retrieveKnowledge(binding, request.getQuery(), request.getLimit(), request.getScoreThreshold()));
registrations.add(new AgentKnowledgeRegistration(spec,
request -> retrieveKnowledge(binding, request.getQuery(), request.getLimit(), request.getScoreThreshold())));
}
definition.setKnowledgeSpecs(specs);
bundle.setKnowledgeRetrievers(retrievers);
bundle.setKnowledgeRegistrations(registrations);
}
/**
* 获取并校验知识库英文运行名。
*
* @param knowledge 知识库发布视图
* @return 合法英文运行名
* @throws BusinessException 英文运行名缺失或非法时抛出
*/
private String requireKnowledgeRuntimeName(DocumentCollection knowledge) {
String runtimeName = knowledge == null ? null : knowledge.getEnglishName();
try {
AgentKnowledgeToolNames.build(runtimeName);
return runtimeName.trim();
} catch (RuntimeException exception) {
String knowledgeName = knowledge == null || knowledge.getTitle() == null
? "未知知识库"
: knowledge.getTitle();
throw new BusinessException(400, 400, "知识库“" + knowledgeName
+ "”的英文名称不能为空,且只能包含字母、数字、下划线和连字符", exception);
}
}
private AgentKnowledgeRetrievalResult retrieveKnowledge(AgentKnowledgeBinding binding, String query, int limit, double scoreThreshold) {

View File

@@ -18,6 +18,7 @@ import java.time.Instant;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.UUID;
@@ -563,7 +564,9 @@ public final class AguiAgentRunOutput implements AgentRunOutput {
}
private static boolean isHiddenToolName(String toolName) {
return "retrieve_knowledge".equalsIgnoreCase(toolName)
String normalizedName = toolName == null ? "" : toolName.trim().toLowerCase(Locale.ROOT);
return "retrieve_knowledge".equals(normalizedName)
|| normalizedName.startsWith("retrieve_knowledge_")
|| "context_reload".equalsIgnoreCase(toolName)
|| "__fragment__".equalsIgnoreCase(toolName);
}

View File

@@ -8,6 +8,7 @@ import org.junit.Test;
import tech.easyflow.agent.entity.Agent;
import tech.easyflow.agent.entity.AgentToolBinding;
import tech.easyflow.agent.enums.AgentToolType;
import tech.easyflow.agent.runtime.skill.AgentSkillRuntimeCompiler;
import tech.easyflow.agent.runtime.tool.AgentToolRuntimeCompiler;
import tech.easyflow.ai.entity.Mcp;
import tech.easyflow.ai.entity.Model;
@@ -44,6 +45,8 @@ public class AgentDefinitionCompilerMcpTest {
setField(toolCompiler, "objectMapper", new com.fasterxml.jackson.databind.ObjectMapper());
setField(toolCompiler, "mcpService", mcpService(mcp));
setField(compiler, "agentToolRuntimeCompiler", toolCompiler);
setField(compiler, "agentSkillRuntimeCompiler", new AgentSkillRuntimeCompiler(
null, toolCompiler, new com.fasterxml.jackson.databind.ObjectMapper()));
Agent agent = agent(modelId, mcpId);

View File

@@ -476,18 +476,48 @@ public class AgentRunServiceDraftAndHitlTest {
}
/**
* 验证知识检索状态不会携带命中文档和内部 metadata
* 验证知识库工具开始事件会投影为脱敏的检索状态。
*
* @throws Exception 反射调用失败时抛出
*/
@Test
public void handleRuntimeEventShouldWhitelistKnowledgeStatusPayload() throws Exception {
public void handleRuntimeEventShouldProjectKnowledgeToolCallAsRunningStatus() throws Exception {
AgentRunService service = new AgentRunService();
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL);
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.TOOL_CALL);
event.setToolCallId("knowledge-call-1");
event.getPayload().put("toolCallId", "knowledge-call-1");
event.getPayload().put("toolName", "retrieve_knowledge_homeinn_faq");
event.getPayload().put("toolCategory", "KNOWLEDGE");
event.getPayload().put("documents", List.of(Map.of("chunkContent", "private chunk")));
event.getPayload().put("metadata", Map.of("sourceUri", "private://document"));
invoke(service, "handleRuntimeEvent",
runtimeEventParameterTypes(),
event, "request-knowledge", legacyOutput(emitter), new StringBuilder(),
new ChatAssistantAccumulator(), chatContext(), new AtomicBoolean(false), false);
Assert.assertEquals(1, emitter.envelopes.size());
@SuppressWarnings("unchecked")
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
Assert.assertEquals(Map.of(
"label", "正在检索知识库",
"status", "running",
"statusKey", "knowledge-retrieval"), payload);
}
/**
* 验证知识库工具结果事件会投影为完成状态。
*
* @throws Exception 反射调用失败时抛出
*/
@Test
public void handleRuntimeEventShouldProjectKnowledgeToolResultAsDoneStatus() throws Exception {
AgentRunService service = new AgentRunService();
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
AgentRuntimeEvent event = knowledgeToolEvent(
AgentRuntimeEventType.TOOL_RESULT, "knowledge-call-1", true);
invoke(service, "handleRuntimeEvent",
runtimeEventParameterTypes(),
event, "request-knowledge", legacyOutput(emitter), new StringBuilder(),
@@ -502,6 +532,48 @@ public class AgentRunServiceDraftAndHitlTest {
"statusKey", "knowledge-retrieval"), payload);
}
/**
* 验证文档摘要事件不会抢先把知识库工具状态标记为完成。
*
* @throws Exception 反射调用失败时抛出
*/
@Test
public void handleRuntimeEventShouldNotCompleteKnowledgeStatusFromDocumentEvent() throws Exception {
AgentRunService service = new AgentRunService();
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.KNOWLEDGE_RETRIEVAL);
event.getPayload().put("documents", List.of(Map.of("chunkContent", "private chunk")));
invoke(service, "handleRuntimeEvent",
runtimeEventParameterTypes(),
event, "request-knowledge", legacyOutput(emitter), new StringBuilder(),
new ChatAssistantAccumulator(), chatContext(), new AtomicBoolean(false), false);
Assert.assertTrue(emitter.envelopes.isEmpty());
}
/**
* 验证并行知识库调用全部结束后才进入终态,并保留任一调用失败结果。
*/
@Test
public void knowledgeStatusTrackerShouldAggregateParallelToolCalls() {
AgentRunService.KnowledgeRetrievalStatusTracker tracker =
new AgentRunService.KnowledgeRetrievalStatusTracker();
AgentRuntimeEvent firstCall = knowledgeToolEvent(
AgentRuntimeEventType.TOOL_CALL, "knowledge-call-1", true);
AgentRuntimeEvent secondCall = knowledgeToolEvent(
AgentRuntimeEventType.TOOL_CALL, "knowledge-call-2", true);
AgentRuntimeEvent firstResult = knowledgeToolEvent(
AgentRuntimeEventType.TOOL_RESULT, "knowledge-call-1", false);
AgentRuntimeEvent secondResult = knowledgeToolEvent(
AgentRuntimeEventType.TOOL_RESULT, "knowledge-call-2", true);
Assert.assertEquals("running", tracker.update(firstCall));
Assert.assertEquals("running", tracker.update(secondCall));
Assert.assertEquals("running", tracker.update(firstResult));
Assert.assertEquals("error", tracker.update(secondResult));
}
/**
* 验证完成事件不会再次发送正文消息,只用于最终收口。
*
@@ -1579,6 +1651,29 @@ public class AgentRunServiceDraftAndHitlTest {
}
}
/**
* 创建知识库工具生命周期测试事件。
*
* @param eventType 工具开始或结果事件类型
* @param toolCallId 工具调用 ID
* @param success 工具结果是否成功
* @return 知识库工具事件
*/
private AgentRuntimeEvent knowledgeToolEvent(AgentRuntimeEventType eventType,
String toolCallId,
boolean success) {
AgentRuntimeEvent event = AgentRuntimeEvent.of(eventType);
event.setToolCallId(toolCallId);
event.getPayload().put("toolCallId", toolCallId);
event.getPayload().put("toolName", "retrieve_knowledge_homeinn_faq");
event.getPayload().put("toolCategory", "KNOWLEDGE");
if (eventType == AgentRuntimeEventType.TOOL_RESULT) {
event.getPayload().put("success", success);
event.getPayload().put("status", success ? "SUCCESS" : "FAILED");
}
return event;
}
private Class<?>[] runtimeEventParameterTypes() {
return new Class<?>[]{AgentRuntimeEvent.class, String.class, AgentRunOutput.class, StringBuilder.class,
ChatAssistantAccumulator.class,

View File

@@ -0,0 +1,268 @@
package tech.easyflow.agent.runtime;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeDocument;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRegistration;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetrievalRequest;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeRetrievalResult;
import com.easyagents.agent.runtime.knowledge.AgentKnowledgeSpec;
import com.easyagents.core.document.Document;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.junit.Assert;
import org.junit.Test;
import tech.easyflow.agent.entity.Agent;
import tech.easyflow.agent.entity.AgentKnowledgeBinding;
import tech.easyflow.agent.runtime.skill.AgentSkillRuntimeCompiler;
import tech.easyflow.agent.runtime.tool.AgentToolRuntimeCompiler;
import tech.easyflow.ai.entity.Model;
import tech.easyflow.ai.entity.ModelProvider;
import tech.easyflow.ai.rag.KnowledgeRetrievalRequest;
import tech.easyflow.ai.service.DocumentCollectionService;
import tech.easyflow.ai.service.ModelService;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.lang.reflect.Field;
import java.lang.reflect.Proxy;
import java.math.BigInteger;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicReference;
/**
* Agent 知识库一库一工具运行时编译测试。
*/
public class AgentRuntimeCompilerKnowledgeTest {
/**
* 验证知识库英文名称、描述和检索配置会编译到中立声明及独立 Retriever。
*
* @throws Exception 反射注入依赖失败时抛出
*/
@Test
public void compileShouldBuildOneKnowledgeRegistrationWithEnglishRuntimeName() throws Exception {
AtomicReference<KnowledgeRetrievalRequest> capturedRequest = new AtomicReference<>();
Document document = new Document("如家酒店通常在入住日 14:00 后办理入住。");
document.setId("chunk-1");
document.setTitle("如家 FAQ");
document.setScore(0.92D);
document.addMetadata("documentId", "faq-document-1");
document.addMetadata("chunkId", "faq-chunk-1");
AgentRuntimeCompiler compiler = compiler(capturedRequest, List.of(document));
Agent agent = agent(knowledgeBinding("homeinn_faq", BigInteger.valueOf(20L)));
AgentRuntimeBundle bundle = compiler.compile(agent);
Assert.assertEquals(1, bundle.getDefinition().getKnowledgeSpecs().size());
AgentKnowledgeSpec spec = bundle.getDefinition().getKnowledgeSpecs().get(0);
Assert.assertEquals("homeinn_faq", spec.getRuntimeName());
Assert.assertEquals("如家 FAQ", spec.getName());
Assert.assertTrue(spec.getDescription().contains("入住"));
Assert.assertEquals(7, spec.getLimit());
Assert.assertEquals(0.55D, spec.getScoreThreshold(), 0.0001D);
Assert.assertEquals(1, bundle.getKnowledgeRegistrations().size());
AgentKnowledgeRegistration registration = bundle.getKnowledgeRegistrations().get(0);
AgentKnowledgeRetrievalRequest retrievalRequest = new AgentKnowledgeRetrievalRequest();
retrievalRequest.setQuery("如家几点入住");
retrievalRequest.setLimit(spec.getLimit());
retrievalRequest.setScoreThreshold(spec.getScoreThreshold());
AgentKnowledgeRetrievalResult result = registration.getRetriever().retrieve(retrievalRequest);
Assert.assertEquals("如家几点入住", capturedRequest.get().getQuery());
Assert.assertEquals(Integer.valueOf(7), capturedRequest.get().getLimit());
Assert.assertEquals(Double.valueOf(0.55D), capturedRequest.get().getMinSimilarity());
Assert.assertEquals("AGENT_KNOWLEDGE", capturedRequest.get().getCallerType());
Assert.assertEquals(1, result.getDocuments().size());
AgentKnowledgeDocument mapped = result.getDocuments().get(0);
Assert.assertEquals("faq-document-1", mapped.getDocumentId());
Assert.assertEquals("faq-chunk-1", mapped.getChunkId());
Assert.assertEquals(0.92D, mapped.getScore(), 0.0001D);
}
/**
* 验证缺失知识库英文名称时在发布编译阶段明确失败。
*
* @throws Exception 反射注入依赖失败时抛出
*/
@Test
public void compileShouldRejectMissingKnowledgeEnglishName() throws Exception {
AgentRuntimeCompiler compiler = compiler(new AtomicReference<>(), List.of());
Agent agent = agent(knowledgeBinding(null, BigInteger.valueOf(20L)));
try {
compiler.compile(agent);
Assert.fail("缺失英文名称时应拒绝编译");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("英文名称不能为空"));
}
}
/**
* 验证多个知识库生成相同工具名时在编译阶段拒绝发布。
*
* @throws Exception 反射注入依赖失败时抛出
*/
@Test
public void compileShouldRejectDuplicateKnowledgeToolNames() throws Exception {
AgentRuntimeCompiler compiler = compiler(new AtomicReference<>(), List.of());
AgentKnowledgeBinding first = knowledgeBinding("homeinn_faq", BigInteger.valueOf(20L));
AgentKnowledgeBinding second = knowledgeBinding("homeinn_faq", BigInteger.valueOf(21L));
Agent agent = agent(first, second);
try {
compiler.compile(agent);
Assert.fail("重复知识库工具名时应拒绝编译");
} catch (BusinessException expected) {
Assert.assertTrue(expected.getMessage().contains("retrieve_knowledge_homeinn_faq"));
}
}
/**
* 创建仅含测试模型与知识库服务的运行时编译器。
*
* @param capturedRequest 检索请求捕获器
* @param documents 检索服务返回文档
* @return 已注入依赖的编译器
* @throws Exception 反射注入失败时抛出
*/
private AgentRuntimeCompiler compiler(AtomicReference<KnowledgeRetrievalRequest> capturedRequest,
List<Document> documents) throws Exception {
ObjectMapper objectMapper = new ObjectMapper();
AgentToolRuntimeCompiler toolCompiler = new AgentToolRuntimeCompiler();
AgentRuntimeCompiler compiler = new AgentRuntimeCompiler();
setField(compiler, "objectMapper", objectMapper);
setField(compiler, "modelService", modelService(model()));
setField(compiler, "documentCollectionService", documentCollectionService(capturedRequest, documents));
setField(compiler, "agentToolRuntimeCompiler", toolCompiler);
setField(compiler, "agentSkillRuntimeCompiler",
new AgentSkillRuntimeCompiler(null, toolCompiler, objectMapper));
return compiler;
}
/**
* 创建带知识库绑定的 Agent。
*
* @param bindings 知识库绑定
* @return Agent 测试对象
*/
private Agent agent(AgentKnowledgeBinding... bindings) {
Agent agent = new Agent();
agent.setId(BigInteger.ONE);
agent.setName("如家助手");
agent.setModelId(BigInteger.TEN);
agent.setKnowledgeBindings(List.of(bindings));
return agent;
}
/**
* 创建冻结知识库绑定。
*
* @param englishName 知识库英文名称
* @param knowledgeId 知识库 ID
* @return 知识库绑定
*/
private AgentKnowledgeBinding knowledgeBinding(String englishName, BigInteger knowledgeId) {
AgentKnowledgeBinding binding = new AgentKnowledgeBinding();
binding.setAgentId(BigInteger.ONE);
binding.setKnowledgeId(knowledgeId);
binding.setRetrievalMode("HYBRID");
binding.setEnabled(true);
binding.setOptionsJson(Map.of("limit", 7, "scoreThreshold", 0.55D));
binding.setResourceSnapshot(Map.of(
"id", knowledgeId,
"title", "如家 FAQ",
"description", "如家酒店入住、退房和会员服务常见问题",
"collectionType", "FAQ",
"englishName", englishName == null ? "" : englishName));
return binding;
}
/**
* 创建模型服务代理。
*
* @param model 测试模型
* @return 模型服务代理
*/
private ModelService modelService(Model model) {
return (ModelService) Proxy.newProxyInstance(
ModelService.class.getClassLoader(),
new Class<?>[]{ModelService.class},
(proxy, method, args) -> "getModelInstance".equals(method.getName())
? model
: defaultValue(method.getReturnType()));
}
/**
* 创建知识库服务代理。
*
* @param capturedRequest 检索请求捕获器
* @param documents 返回文档
* @return 知识库服务代理
*/
private DocumentCollectionService documentCollectionService(
AtomicReference<KnowledgeRetrievalRequest> capturedRequest,
List<Document> documents) {
return (DocumentCollectionService) Proxy.newProxyInstance(
DocumentCollectionService.class.getClassLoader(),
new Class<?>[]{DocumentCollectionService.class},
(proxy, method, args) -> {
if ("search".equals(method.getName()) && args != null && args.length == 1
&& args[0] instanceof KnowledgeRetrievalRequest request) {
capturedRequest.set(request);
return documents;
}
return defaultValue(method.getReturnType());
});
}
/**
* 创建可映射为 AgentScope 模型配置的测试模型。
*
* @return 测试模型
*/
private Model model() {
ModelProvider provider = new ModelProvider();
provider.setProviderType("openai");
provider.setProviderName("OpenAI");
Model model = new Model();
model.setId(BigInteger.TEN);
model.setModelProvider(provider);
model.setModelName("gpt-test");
model.setEndpoint("https://example.com");
model.setRequestPath("/v1/chat/completions");
model.setApiKey("test-key");
return model;
}
/**
* 返回代理方法所需的默认值。
*
* @param type 返回类型
* @return 对应默认值
*/
private Object defaultValue(Class<?> type) {
if (type == boolean.class) {
return false;
}
if (type == int.class || type == long.class || type == short.class || type == byte.class) {
return 0;
}
if (type == double.class || type == float.class) {
return 0D;
}
return null;
}
/**
* 反射注入测试依赖。
*
* @param target 目标对象
* @param fieldName 字段名称
* @param value 字段值
* @throws Exception 字段不存在或不可写时抛出
*/
private void setField(Object target, String fieldName, Object value) throws Exception {
Field field = target.getClass().getDeclaredField(fieldName);
field.setAccessible(true);
field.set(target, value);
}
}

View File

@@ -10,6 +10,7 @@ import tech.easyflow.ai.document.exception.DocumentParseBridgeException;
import tech.easyflow.ai.document.model.DocumentSourceRef;
import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadedFileReader;
import tech.easyflow.common.filestorage.FileStorageService;
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
import tech.easyflow.common.filestorage.utils.PathGeneratorUtil;
import java.io.IOException;
@@ -164,7 +165,7 @@ public class DocumentSourceLoader {
}
/**
* 优先打开经过上传记录验证的受管 URL再执行普通公网 URL 校验与下载。
* 优先打开经过上传记录验证的受管 URL 和服务端已登记附件,再执行普通公网 URL 校验与下载。
*
* @param remoteUrl 远端 URL
* @param maxBytes 最大允许读取字节数
@@ -176,6 +177,13 @@ public class DocumentSourceLoader {
if (managed.isPresent()) {
return DocumentInputStreamSupport.limit(managed.get(), maxBytes);
}
Optional<FileStorageWriteHandle> trusted =
fileStorageService.resolveTrustedFile(remoteUrl);
if (trusted.isPresent()) {
return DocumentInputStreamSupport.limit(
fileStorageService.readRecoverable(trusted.get()),
maxBytes);
}
return DocumentInputStreamSupport.openRemote(remoteUrl, maxBytes);
}

View File

@@ -15,6 +15,7 @@ import tech.easyflow.ai.document.support.DocumentInputStreamSupport;
import tech.easyflow.ai.document.support.DocumentParseSourceType;
import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadedFileReader;
import tech.easyflow.common.filestorage.FileStorageService;
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
import tech.easyflow.common.util.StringUtil;
import tech.easyflow.common.web.exceptions.BusinessException;
@@ -326,20 +327,29 @@ public class DocNodeFileContentExtractor {
private void copySourceToTemporaryFile(
DocumentSourceRef sourceRef, Path target) throws IOException {
String filePath = sourceRef.getFilePath();
boolean managedUpload = StringUtil.hasText(filePath)
&& uploadedFileReader != null
&& uploadedFileReader.isManagedPathCandidate(filePath);
Optional<FileStorageWriteHandle> trustedFile = Optional.empty();
if (StringUtil.hasText(filePath)
&& isRemoteUrl(filePath)
&& !managedUpload) {
trustedFile = fileStorageService.resolveTrustedFile(filePath);
}
boolean localStorage = StringUtil.hasText(filePath)
&& (!isRemoteUrl(filePath)
|| (uploadedFileReader != null
&& uploadedFileReader.isManagedPathCandidate(filePath)));
|| managedUpload
|| trustedFile.isPresent());
if (localStorage) {
try (IoBulkhead.Permit ignored =
IoBulkhead.storage().acquire("storage:document-read");
InputStream inputStream = openInputStream(sourceRef);
InputStream inputStream = openInputStream(sourceRef, trustedFile);
OutputStream outputStream = Files.newOutputStream(target)) {
copy(inputStream, outputStream);
}
return;
}
try (InputStream inputStream = openInputStream(sourceRef);
try (InputStream inputStream = openInputStream(sourceRef, trustedFile);
OutputStream outputStream = Files.newOutputStream(target)) {
copy(inputStream, outputStream);
}
@@ -362,7 +372,17 @@ public class DocNodeFileContentExtractor {
}
}
private InputStream openInputStream(DocumentSourceRef sourceRef) throws IOException {
/**
* 按可信受管上传、服务端文件记录、本地路径和普通公网 URL 的顺序打开源流。
*
* @param sourceRef 文档源
* @param trustedFile 已通过服务端记录或存储配置确认的物理读取句柄
* @return 受实际字节数限制的输入流
* @throws IOException 文件无法安全读取时抛出
*/
private InputStream openInputStream(
DocumentSourceRef sourceRef,
Optional<FileStorageWriteHandle> trustedFile) throws IOException {
String filePath = sourceRef.getFilePath();
if (uploadedFileReader != null && StringUtil.hasText(filePath)) {
Optional<InputStream> managed = uploadedFileReader.openVerified(filePath);
@@ -372,6 +392,11 @@ public class DocNodeFileContentExtractor {
FILE_MAX_SINGLE_SIZE);
}
}
if (trustedFile.isPresent()) {
return DocumentInputStreamSupport.limit(
fileStorageService.readRecoverable(trustedFile.get()),
FILE_MAX_SINGLE_SIZE);
}
if (StringUtil.hasText(filePath) && isRemoteUrl(filePath)) {
return DocumentInputStreamSupport.openRemote(filePath, FILE_MAX_SINGLE_SIZE);
}

View File

@@ -187,7 +187,7 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
RagScoreNormalizer.normalize(searchDocuments, retrievalMode, reranked);
List<Document> formattedDocuments = formatDocuments(
searchDocuments,
shouldApplyMinSimilarityFilter(retrievalMode, reranked),
true,
minSimilarity,
docRecallMaxNum
);
@@ -396,10 +396,6 @@ public class DocumentCollectionServiceImpl extends ServiceImpl<DocumentCollectio
return modelRerank.toRerankModel();
}
private boolean shouldApplyMinSimilarityFilter(RetrievalMode retrievalMode, boolean reranked) {
return !reranked && retrievalMode == RetrievalMode.VECTOR;
}
/**
* 解析本次查询使用的召回上限,优先采用请求参数,其次回退到知识库默认配置。
*

View File

@@ -7,6 +7,7 @@ import tech.easyflow.ai.easyagentsflow.upload.WorkflowApiUploadedFileReader;
import tech.easyflow.ai.document.exception.DocumentParseBridgeException;
import tech.easyflow.ai.document.model.DocumentSourceRef;
import tech.easyflow.common.filestorage.FileStorageService;
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
import java.io.ByteArrayInputStream;
import java.io.IOException;
@@ -73,6 +74,24 @@ public class DocumentSourceLoaderTest {
}
}
/**
* 验证服务端已登记的普通附件 URL 在公网地址校验前通过物理句柄直读。
*/
@Test
public void shouldLoadRecordedInternalStorageUrlBeforeRemoteAddressGuard() {
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/1/2026/8/26/demo.pdf";
byte[] body = "recorded-pdf".getBytes(StandardCharsets.UTF_8);
DocumentSourceLoader loader = new DocumentSourceLoader(
new RecordedFileStorageService(fileUrl, body));
DocumentSourceRef sourceRef = new DocumentSourceRef();
sourceRef.setFileName("demo.pdf");
sourceRef.setFilePath(fileUrl);
LoadedDocumentSource loadedSource = loader.load(sourceRef);
Assert.assertArrayEquals(body, loadedSource.getContentBytes());
}
/**
* 验证已通过上传记录校验的内网存储 URL 会走恢复句柄读取。
*
@@ -217,4 +236,42 @@ public class DocumentSourceLoaderTest {
return 0L;
}
}
/**
* 仅允许通过服务端记录句柄读取内容的存储测试替身。
*/
private static class RecordedFileStorageService
extends FailingFileStorageService {
/** 允许解析的精确 URL。 */
private final String recordedUrl;
/** 固定文件内容。 */
private final byte[] content;
/** 固定可信读取句柄。 */
private final FileStorageWriteHandle handle = new FileStorageWriteHandle(
"recorded", "", "/storage", "attachment", "demo.pdf");
/**
* 创建服务端记录存储替身。
*
* @param recordedUrl 允许解析的精确 URL
* @param content 固定文件内容
*/
private RecordedFileStorageService(String recordedUrl, byte[] content) {
this.recordedUrl = recordedUrl;
this.content = content.clone();
}
/** {@inheritDoc} */
@Override
public Optional<FileStorageWriteHandle> resolveTrustedFile(String reference) {
return recordedUrl.equals(reference) ? Optional.of(handle) : Optional.empty();
}
/** {@inheritDoc} */
@Override
public InputStream readRecoverable(FileStorageWriteHandle requestedHandle) {
Assert.assertSame(handle, requestedHandle);
return new ByteArrayInputStream(content);
}
}
}

View File

@@ -10,6 +10,7 @@ import tech.easyflow.ai.document.model.DocumentParsedResult;
import tech.easyflow.ai.document.model.DocumentSourceRef;
import tech.easyflow.ai.document.service.DocumentParseBridgeService;
import tech.easyflow.common.filestorage.FileStorageService;
import tech.easyflow.common.filestorage.FileStorageWriteHandle;
import tech.easyflow.common.web.exceptions.BusinessException;
import java.io.ByteArrayInputStream;
@@ -203,6 +204,27 @@ public class DocNodeFileContentExtractorTest {
Assert.assertNull(bridgeService.lastSource);
}
/**
* 验证默认读取器也能通过服务端文件记录读取内部附件 URL。
*/
@Test
public void shouldReadRecordedInternalUrlForUnsupportedType() {
RecordingDocumentParseBridgeService bridgeService = new RecordingDocumentParseBridgeService();
String fileUrl = "http://127.0.0.1:39000/easyflow/attachment/1/2026/8/26/note.txt";
DocNodeFileContentExtractor extractor = new DocNodeFileContentExtractor(
bridgeService,
new RecordedFileStorageService(fileUrl, "recorded text"),
new ReadingReaderManager());
String content = extractor.extract(buildFileValue(
"note.txt",
fileUrl,
"text/plain"));
Assert.assertEquals("recorded text", content);
Assert.assertNull(bridgeService.lastSource);
}
/**
* 验证受管上传 URL 的非桥接文件通过记录校验后走内部存储读取。
*
@@ -574,4 +596,42 @@ public class DocNodeFileContentExtractorTest {
return 0L;
}
}
/**
* 仅允许通过服务端记录句柄读取内容的存储测试替身。
*/
private static class RecordedFileStorageService
extends FailingFileStorageService {
/** 允许解析的精确 URL。 */
private final String recordedUrl;
/** 固定内容。 */
private final byte[] content;
/** 固定可信读取句柄。 */
private final FileStorageWriteHandle handle = new FileStorageWriteHandle(
"recorded", "", "/storage", "attachment", "note.txt");
/**
* 创建服务端记录存储替身。
*
* @param recordedUrl 允许解析的精确 URL
* @param content 固定文本内容
*/
private RecordedFileStorageService(String recordedUrl, String content) {
this.recordedUrl = recordedUrl;
this.content = content.getBytes(StandardCharsets.UTF_8);
}
/** {@inheritDoc} */
@Override
public Optional<FileStorageWriteHandle> resolveTrustedFile(String reference) {
return recordedUrl.equals(reference) ? Optional.of(handle) : Optional.empty();
}
/** {@inheritDoc} */
@Override
public InputStream readRecoverable(FileStorageWriteHandle requestedHandle) {
Assert.assertSame(handle, requestedHandle);
return new ByteArrayInputStream(content);
}
}
}

View File

@@ -32,6 +32,28 @@ import static tech.easyflow.ai.entity.DocumentCollection.KEY_SIMILARITY_THRESHOL
*/
public class DocumentCollectionServiceImplTest {
/**
* 验证最终相关度阈值会过滤所有已统一到零到一范围的检索结果。
*/
@Test
public void formatDocumentsShouldApplyFinalScoreThreshold() {
Document lowScore = buildHit(BigInteger.ONE, 0.49D);
Document thresholdScore = buildHit(BigInteger.TWO, 0.5D);
Document highScore = buildHit(BigInteger.valueOf(3), 0.9D);
DocumentCollectionServiceImpl service = new DocumentCollectionServiceImpl();
List<Document> result = service.formatDocuments(
List.of(lowScore, thresholdScore, highScore),
true,
0.5F,
5
);
Assert.assertEquals(2, result.size());
Assert.assertEquals(highScore.getId(), result.get(0).getId());
Assert.assertEquals(thresholdScore.getId(), result.get(1).getId());
}
/**
* 验证检索结果会在重排前过滤掉未完成文档,避免高分进行中文档挤占最终名额。
*

View File

@@ -2,6 +2,7 @@ import { EventType } from '@ag-ui/client';
import { afterEach, describe, expect, it, vi } from 'vitest';
import { EasyFlowAguiClient, EasyFlowAguiProjectionError } from './client';
import { easyFlowAguiCustomEvent } from './custom-events';
import { isRetryableAguiTransportError } from './reconnect';
vi.mock('#/api/request', () => ({
@@ -276,6 +277,69 @@ describe('easyFlowAguiClient', () => {
expect(received.at(-1)).toBe(EventType.RUN_FINISHED);
});
it('yields a paint opportunity after knowledge retrieval starts', async () => {
vi.useFakeTimers();
let paintCallback: FrameRequestCallback | undefined;
vi.stubGlobal(
'requestAnimationFrame',
vi.fn((callback: FrameRequestCallback) => {
paintCallback = callback;
return 1;
}),
);
vi.stubGlobal(
'fetch',
vi.fn(async () =>
sse([
{ runId: 'run-1', threadId: '101', type: EventType.RUN_STARTED },
{
name: easyFlowAguiCustomEvent.knowledgeRetrievalStatus,
type: EventType.CUSTOM,
value: {
status: 'running',
statusKey: 'knowledge-retrieval',
},
},
{
name: easyFlowAguiCustomEvent.knowledgeRetrievalStatus,
type: EventType.CUSTOM,
value: {
status: 'done',
statusKey: 'knowledge-retrieval',
},
},
{
runId: 'run-1',
threadId: '101',
type: EventType.RUN_FINISHED,
},
]),
),
);
const receivedStatuses: string[] = [];
const runPromise = new EasyFlowAguiClient().run({
onEvent: (event) => {
if (event.type === EventType.CUSTOM) {
const value = event.value as Record<string, unknown>;
receivedStatuses.push(String(value.status));
}
},
threadId: '101',
url: '/api/v1/agent/1/agui/run',
userMessage: { content: '几点退房', id: 'user-1', role: 'user' },
});
await vi.advanceTimersByTimeAsync(0);
expect(receivedStatuses).toEqual(['running']);
paintCallback?.(0);
await vi.advanceTimersByTimeAsync(0);
await runPromise;
expect(receivedStatuses).toEqual(['running', 'done']);
});
it('replays a completed run from the server journal after refresh', async () => {
vi.stubGlobal(
'fetch',

View File

@@ -5,6 +5,8 @@ import { events } from 'fetch-event-stream';
import { createEventStreamHeaders, resolveApiUrl } from '#/api/request';
import { easyFlowAguiCustomEvent } from './custom-events';
export interface EasyFlowAguiRunOptions {
forwardedProps?: Record<string, unknown>;
onCursor?: (cursor: number) => void;
@@ -101,6 +103,31 @@ function waitForToolStartPaint(): Promise<void> {
});
}
/**
* 判断事件是否开启了需要即时呈现的工具执行状态。
*
* @param event AG-UI 事件
* @returns 标准工具开始或知识库检索开始时为 true
*/
function startsVisibleToolExecution(event: AguiEvent) {
if (event.type === EventType.TOOL_CALL_START) {
return true;
}
if (
event.type !== EventType.CUSTOM ||
event.name !== easyFlowAguiCustomEvent.knowledgeRetrievalStatus
) {
return false;
}
const value =
event.value &&
typeof event.value === 'object' &&
!Array.isArray(event.value)
? (event.value as Record<string, unknown>)
: {};
return String(value.status || '').toLowerCase() === 'running';
}
/**
* EasyFlow 的无头 AG-UI 运行客户端。
*
@@ -174,7 +201,7 @@ export class EasyFlowAguiClient {
) {
terminalReceived = true;
}
if (event.type === EventType.TOOL_CALL_START) {
if (startsVisibleToolExecution(event as AguiEvent)) {
await waitForToolStartPaint();
}
}
@@ -233,7 +260,7 @@ export class EasyFlowAguiClient {
if (Number.isSafeInteger(cursor) && cursor > 0) {
options.onCursor?.(cursor);
}
if (event.type === EventType.TOOL_CALL_START) {
if (startsVisibleToolExecution(event as AguiEvent)) {
await waitForToolStartPaint();
}
}

View File

@@ -336,6 +336,76 @@ describe('aG-UI wire contract and timeline projection', () => {
).toBe(true);
});
it('merges knowledge retrieval tool and status events within one turn', () => {
const items: ChatTimelineItem[] = [];
const state = createAguiTimelineProjectionState();
const events = [
{
toolCallId: 'tool-faq-1',
toolCallName: 'retrieve_knowledge_homeinn_faq',
type: EventType.TOOL_CALL_START,
},
{
name: easyFlowAguiCustomEvent.knowledgeRetrievalStatus,
type: EventType.CUSTOM,
value: {
label: '已检索知识库',
status: 'done',
statusKey: 'knowledge-retrieval',
},
},
{
content: 'Retrieved 1 relevant document(s)',
messageId: 'tool-result-faq-1',
role: 'tool',
toolCallId: 'tool-faq-1',
type: EventType.TOOL_CALL_RESULT,
},
].map((event) => EventSchemas.parse(event));
for (const event of events) {
applyAguiEventToTimeline(items, event, { roundId: 'round-faq' }, state);
}
expect(items).toHaveLength(1);
expect(items[0]).toMatchObject({
label: '已检索知识库',
roundId: 'round-faq',
status: 'done',
statusKey: 'knowledge-retrieval:round-faq',
type: 'status',
});
});
it('projects failed knowledge retrieval without exposing tool details', () => {
const items: ChatTimelineItem[] = [];
applyAguiEventToTimeline(
items,
EventSchemas.parse({
name: easyFlowAguiCustomEvent.knowledgeRetrievalStatus,
type: EventType.CUSTOM,
value: {
internalError: 'private stack',
status: 'error',
statusKey: 'knowledge-retrieval',
},
}),
{ roundId: 'round-failed-knowledge' },
createAguiTimelineProjectionState(),
);
expect(items).toEqual([
expect.objectContaining({
label: '知识库检索失败',
status: 'error',
statusKey: 'knowledge-retrieval:round-failed-knowledge',
tone: 'danger',
type: 'status',
}),
]);
expect(JSON.stringify(items)).not.toContain('private stack');
});
it('projects Skill invocation status in place through the strict public fields', () => {
const items: ChatTimelineItem[] = [];
const state = createAguiTimelineProjectionState();

View File

@@ -3,6 +3,7 @@ import type {
ChatTimelineKnowledgeHit,
ChatTimelineMessageItem,
ChatTimelineSkillInvocationStatus,
ChatTimelineStatusStatus,
ChatTimelineToolStatus,
} from '@easyflow/common-ui';
@@ -131,6 +132,13 @@ function asyncToolStatus(
return 'running';
}
function knowledgeRetrievalStatus(value: unknown): ChatTimelineStatusStatus {
const status = asText(value).trim().toLowerCase();
if (status === 'running') return 'running';
if (status === 'error' || status === 'failed') return 'error';
return 'done';
}
function statusKey(
payload: Record<string, unknown>,
options: AguiTimelineProjectionOptions,
@@ -275,7 +283,7 @@ function applyCustomEvent(
if (event.name === easyFlowAguiCustomEvent.knowledgeRetrievalStatus) {
ChatTimelineBuilder.upsertKnowledgeRetrievalStatus(
items,
asText(payload.status).toLowerCase() === 'running' ? 'running' : 'done',
knowledgeRetrievalStatus(payload.status),
statusKey(payload, options, 'knowledge-retrieval'),
turnMetadata,
);

View File

@@ -54,20 +54,29 @@ function isHiddenToolName(toolName?: string) {
const normalizedName = normalizeToolName(toolName);
return (
normalizedName === 'retrieve_knowledge' ||
normalizedName.startsWith('retrieve_knowledge_') ||
normalizedName === 'context_reload' ||
normalizedName === '__fragment__'
);
}
function isKnowledgeRetrievalToolName(toolName?: string) {
return normalizeToolName(toolName) === 'retrieve_knowledge';
const normalizedName = normalizeToolName(toolName);
return (
normalizedName === 'retrieve_knowledge' ||
normalizedName.startsWith('retrieve_knowledge_')
);
}
function isBlankToolName(toolName?: string) {
return !normalizeToolName(toolName);
}
function knowledgeRetrievalStatusKey(statusKey?: string) {
function knowledgeRetrievalStatusKey(statusKey?: string, roundId?: string) {
const normalizedRoundId = normalizeText(roundId).trim();
if (normalizedRoundId) {
return `knowledge-retrieval:${normalizedRoundId}`;
}
return normalizeText(statusKey).trim() || 'knowledge-retrieval';
}
@@ -659,12 +668,18 @@ export const ChatTimelineBuilder = {
metadata?: ChatTimelineTurnMetadata,
) {
finishAssistantMessage(items, false, metadata?.roundId);
let label = '已检索知识库';
if (status === 'running') {
label = '正在检索知识库';
} else if (status === 'error') {
label = '知识库检索失败';
}
upsertStatus(items, {
...metadata,
label: status === 'running' ? '正在检索知识库' : '已检索知识库',
label,
status,
statusKey: knowledgeRetrievalStatusKey(statusKey),
tone: 'muted',
statusKey: knowledgeRetrievalStatusKey(statusKey, metadata?.roundId),
tone: status === 'error' ? 'danger' : 'muted',
});
},