feat: 全新智能体功能
- 基于先进智能体框架，增加智能体编排功能 - 增加智能体聊天，并对接持久化
This commit is contained in:
@@ -0,0 +1,59 @@
|
||||
package tech.easyflow.agent.publish;
|
||||
|
||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.agent.service.AgentKnowledgeBindingService;
|
||||
import tech.easyflow.agent.service.AgentToolBindingService;
|
||||
|
||||
import java.lang.reflect.Proxy;
|
||||
import java.math.BigInteger;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
/**
|
||||
* {@link AgentApprovalSubjectHandler} 单元测试。
|
||||
*/
|
||||
public class AgentApprovalSubjectHandlerTest {
|
||||
|
||||
/**
|
||||
* 审批删除 Agent 前必须同步清理工具绑定和知识库绑定，避免留下孤儿数据。
|
||||
*/
|
||||
@Test
|
||||
public void beforeRemoveShouldCleanAgentBindings() {
|
||||
AtomicInteger toolRemoveCalls = new AtomicInteger();
|
||||
AtomicInteger knowledgeRemoveCalls = new AtomicInteger();
|
||||
AgentToolBindingService toolBindingService = proxy(AgentToolBindingService.class, toolRemoveCalls);
|
||||
AgentKnowledgeBindingService knowledgeBindingService = proxy(AgentKnowledgeBindingService.class, knowledgeRemoveCalls);
|
||||
AgentApprovalSubjectHandler handler = new AgentApprovalSubjectHandler(
|
||||
null,
|
||||
new ObjectMapper(),
|
||||
null,
|
||||
toolBindingService,
|
||||
knowledgeBindingService,
|
||||
null
|
||||
);
|
||||
|
||||
handler.beforeRemove(BigInteger.valueOf(1001));
|
||||
|
||||
Assert.assertEquals(1, toolRemoveCalls.get());
|
||||
Assert.assertEquals(1, knowledgeRemoveCalls.get());
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private static <T> T proxy(Class<T> type, AtomicInteger removeCalls) {
|
||||
return (T) Proxy.newProxyInstance(
|
||||
type.getClassLoader(),
|
||||
new Class<?>[]{type},
|
||||
(proxy, method, args) -> {
|
||||
if ("remove".equals(method.getName()) && args != null && args.length == 1) {
|
||||
removeCalls.incrementAndGet();
|
||||
return true;
|
||||
}
|
||||
if (method.getReturnType() == boolean.class || method.getReturnType() == Boolean.class) {
|
||||
return false;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,192 @@
|
||||
package tech.easyflow.agent.runtime;
|
||||
|
||||
import com.easyagents.agent.runtime.AgentInitRequest;
|
||||
import com.easyagents.agent.runtime.AgentResumeRequest;
|
||||
import com.easyagents.agent.runtime.AgentRuntime;
|
||||
import com.easyagents.agent.runtime.event.AgentRuntimeEvent;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import reactor.core.publisher.Flux;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.core.runtime.ChatAssistantAccumulator;
|
||||
import tech.easyflow.core.runtime.ChatRuntimeContext;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
/**
|
||||
* Agent 运行态注册表测试。
|
||||
*/
|
||||
public class AgentRunRegistryTest {
|
||||
|
||||
/**
|
||||
* 验证批准请求会恢复当前运行时。
|
||||
*/
|
||||
@Test
|
||||
public void approveShouldResumeRuntime() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
CapturingRuntime runtime = new CapturingRuntime();
|
||||
|
||||
registry.register(context("request-1", "session-1", "user-1", runtime));
|
||||
registry.registerResumeToken("request-1", "token-1");
|
||||
registry.approve("request-1", "token-1", "user-1");
|
||||
|
||||
AgentResumeRequest request = runtime.resumeRequest.get();
|
||||
Assert.assertNotNull(request);
|
||||
Assert.assertTrue(request.isApproved());
|
||||
Assert.assertEquals("token-1", request.getResumeToken().getValue());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证拒绝请求支持通过恢复令牌反查运行时。
|
||||
*/
|
||||
@Test
|
||||
public void rejectShouldResolveRequestByResumeToken() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
CapturingRuntime runtime = new CapturingRuntime();
|
||||
|
||||
registry.register(context("request-2", "session-2", "user-1", runtime));
|
||||
registry.registerResumeToken("request-2", "token-2");
|
||||
registry.reject(null, "token-2", "user-1", "denied");
|
||||
|
||||
AgentResumeRequest request = runtime.resumeRequest.get();
|
||||
Assert.assertNotNull(request);
|
||||
Assert.assertFalse(request.isApproved());
|
||||
Assert.assertEquals("denied", request.getRejectReason());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证运行结束后清理运行态与恢复令牌索引。
|
||||
*/
|
||||
@Test
|
||||
public void removeShouldClearRuntimeAndResumeTokens() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
|
||||
registry.register(context("request-3", "session-3", "user-1", new CapturingRuntime()));
|
||||
registry.registerResumeToken("request-3", "token-3");
|
||||
registry.remove("request-3");
|
||||
|
||||
Assert.assertThrows(BusinessException.class, () -> registry.approve(null, "token-3", "user-1"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证运行审批只能由运行发起人处理。
|
||||
*/
|
||||
@Test
|
||||
public void approveShouldRejectDifferentOwner() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
CapturingRuntime runtime = new CapturingRuntime();
|
||||
|
||||
registry.register(context("request-4", "session-4", "user-1", runtime));
|
||||
registry.registerResumeToken("request-4", "token-4");
|
||||
|
||||
Assert.assertThrows(BusinessException.class, () -> registry.approve("request-4", "token-4", "user-2"));
|
||||
Assert.assertNull(runtime.resumeRequest.get());
|
||||
|
||||
registry.approve("request-4", "token-4", "user-1");
|
||||
Assert.assertTrue(runtime.resumeRequest.get().isApproved());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证传入 requestId 时仍会校验恢复令牌归属，避免错误令牌打断挂起运行。
|
||||
*/
|
||||
@Test
|
||||
public void approveShouldRejectTokenNotBelongingToRequest() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
CapturingRuntime runtime = new CapturingRuntime();
|
||||
|
||||
registry.register(context("request-5", "session-5", "user-1", runtime));
|
||||
|
||||
Assert.assertThrows(BusinessException.class,
|
||||
() -> registry.approve("request-5", "wrong-token", "user-1"));
|
||||
Assert.assertNull(runtime.resumeRequest.get());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证同一会话同一时刻只允许一个运行态。
|
||||
*/
|
||||
@Test
|
||||
public void registerShouldRejectActiveRunInSameSession() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
|
||||
registry.register(context("request-6", "session-6", "user-1", new CapturingRuntime()));
|
||||
|
||||
Assert.assertThrows(BusinessException.class,
|
||||
() -> registry.register(context("request-7", "session-6", "user-1", new CapturingRuntime())));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证按会话清理会取消当前运行并释放同会话运行锁。
|
||||
*/
|
||||
@Test
|
||||
public void cancelSessionShouldRemoveActiveRun() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
AgentRunRegistry.AgentRunContext context = context("request-8", "session-8", "user-1", new CapturingRuntime());
|
||||
|
||||
registry.register(context);
|
||||
registry.cancelSession("session-8");
|
||||
|
||||
Assert.assertNull(registry.get("request-8"));
|
||||
registry.register(context("request-9", "session-8", "user-1", new CapturingRuntime()));
|
||||
Assert.assertNotNull(registry.get("request-9"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证按会话清理时会校验运行归属。
|
||||
*/
|
||||
@Test
|
||||
public void cancelSessionShouldRejectDifferentOwner() {
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
|
||||
registry.register(context("request-10", "session-10", "user-1", new CapturingRuntime()));
|
||||
|
||||
Assert.assertThrows(BusinessException.class, () -> registry.cancelSession("session-10", "user-2"));
|
||||
Assert.assertNotNull(registry.get("request-10"));
|
||||
}
|
||||
|
||||
private AgentRunRegistry.AgentRunContext context(String requestId,
|
||||
String sessionId,
|
||||
String userId,
|
||||
AgentRuntime runtime) {
|
||||
return new AgentRunRegistry.AgentRunContext(
|
||||
requestId,
|
||||
sessionId,
|
||||
runtime,
|
||||
null,
|
||||
new ChatRuntimeContext(),
|
||||
new StringBuilder(),
|
||||
new ChatAssistantAccumulator(),
|
||||
new AtomicBoolean(false),
|
||||
false,
|
||||
new AgentRunRegistry.RunOwner("agent-1", sessionId, userId),
|
||||
null,
|
||||
event -> {
|
||||
},
|
||||
error -> {
|
||||
},
|
||||
() -> {
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
private static final class CapturingRuntime implements AgentRuntime {
|
||||
|
||||
private final AtomicReference<AgentResumeRequest> resumeRequest = new AtomicReference<>();
|
||||
|
||||
@Override
|
||||
public void init(AgentInitRequest request) {
|
||||
// 测试桩无需初始化。
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AgentRuntimeEvent> stream(com.easyagents.agent.runtime.message.AgentMessage userMessage) {
|
||||
return Flux.empty();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Flux<AgentRuntimeEvent> resume(AgentResumeRequest request) {
|
||||
resumeRequest.set(request);
|
||||
return Flux.empty();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,660 @@
|
||||
package tech.easyflow.agent.runtime;
|
||||
|
||||
import com.easyagents.agent.runtime.event.AgentRuntimeEvent;
|
||||
import com.easyagents.agent.runtime.event.AgentRuntimeEventType;
|
||||
import com.easyagents.agent.runtime.message.AgentKnowledgeReference;
|
||||
import com.easyagents.agent.runtime.message.AgentMessage;
|
||||
import com.easyagents.agent.runtime.message.AgentMessageRole;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import tech.easyflow.agent.entity.Agent;
|
||||
import tech.easyflow.agent.entity.AgentKnowledgeBinding;
|
||||
import tech.easyflow.agent.entity.AgentToolBinding;
|
||||
import tech.easyflow.agent.runtime.lock.AgentRunLock;
|
||||
import tech.easyflow.chatlog.domain.dto.ChatSessionSummary;
|
||||
import tech.easyflow.common.entity.LoginAccount;
|
||||
import tech.easyflow.common.web.exceptions.BusinessException;
|
||||
import tech.easyflow.core.chat.protocol.ChatDomain;
|
||||
import tech.easyflow.core.chat.protocol.ChatEnvelope;
|
||||
import tech.easyflow.core.chat.protocol.ChatType;
|
||||
import tech.easyflow.core.chat.protocol.sse.ChatSseEmitter;
|
||||
import tech.easyflow.core.runtime.ChatAssistantAccumulator;
|
||||
import tech.easyflow.core.runtime.ChatRuntimeContext;
|
||||
import tech.easyflow.core.runtime.ChatRuntimeManager;
|
||||
import tech.easyflow.core.runtime.ChatRuntimeMessage;
|
||||
|
||||
import java.lang.reflect.Method;
|
||||
import java.math.BigInteger;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
/**
|
||||
* Agent 草稿试用与 HITL 事件映射测试。
|
||||
*/
|
||||
public class AgentRunServiceDraftAndHitlTest {
|
||||
|
||||
/**
|
||||
* 验证工具 HITL 事件会映射为显式前端载荷。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void buildToolHitlPayloadShouldExposeStableFieldsWithoutPrompt() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.TOOL_APPROVAL_REQUIRED);
|
||||
event.setToolCallId("call-1");
|
||||
event.getPayload().put("resumeToken", "token-1");
|
||||
event.getPayload().put("sessionId", "session-1");
|
||||
event.getPayload().put("agentId", "agent-1");
|
||||
event.getPayload().put("toolName", "search");
|
||||
event.getPayload().put("toolType", "PLUGIN");
|
||||
event.getPayload().put("approvalPrompt", "不应透出");
|
||||
event.getPayload().put("toolInput", Map.of("keyword", "EasyFlow"));
|
||||
event.getPayload().put("approvalMetadata", Map.of(
|
||||
"risk", "low",
|
||||
"prompt", "不应透出",
|
||||
"toolType", "WORKFLOW"
|
||||
));
|
||||
|
||||
AgentToolHitlPayload payload = invoke(service, "buildToolHitlPayload",
|
||||
new Class<?>[]{String.class, AgentRuntimeEvent.class}, "request-1", event);
|
||||
|
||||
Assert.assertEquals("request-1", payload.getRequestId());
|
||||
Assert.assertEquals("token-1", payload.getResumeToken());
|
||||
Assert.assertEquals("session-1", payload.getSessionId());
|
||||
Assert.assertEquals("agent-1", payload.getAgentId());
|
||||
Assert.assertEquals("call-1", payload.getToolCallId());
|
||||
Assert.assertEquals("search", payload.getToolName());
|
||||
Assert.assertEquals("PLUGIN", payload.getToolType());
|
||||
Assert.assertEquals("EasyFlow", payload.getInput().get("keyword"));
|
||||
Assert.assertEquals("low", payload.getMetadata().get("risk"));
|
||||
Assert.assertEquals("PLUGIN", payload.getMetadata().get("toolType"));
|
||||
Assert.assertFalse(payload.getMetadata().containsKey("prompt"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证工具事件发送给前端时会携带稳定工具调用 ID。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void buildToolEventPayloadShouldExposeRuntimeToolCallId() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.TOOL_RESULT);
|
||||
event.setToolCallId("call-runtime");
|
||||
event.getPayload().put("toolName", "search");
|
||||
event.getPayload().put("text", "ok");
|
||||
|
||||
Map<String, Object> payload = invoke(service, "buildToolEventPayload",
|
||||
new Class<?>[]{AgentRuntimeEvent.class}, event);
|
||||
|
||||
Assert.assertEquals("call-runtime", payload.get("toolCallId"));
|
||||
Assert.assertEquals("search", payload.get("toolName"));
|
||||
Assert.assertEquals("ok", payload.get("text"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证思考事件会优先读取 reasoning 字段。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void stringPayloadShouldExposeReasoningValue() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.REASONING_DELTA);
|
||||
event.getPayload().put("reasoning", "思考中");
|
||||
|
||||
String reasoning = invoke(service, "stringPayload",
|
||||
new Class<?>[]{AgentRuntimeEvent.class, String.class}, event, "reasoning");
|
||||
String fallback = invoke(service, "firstText",
|
||||
new Class<?>[]{String.class, String.class}, reasoning, "正文");
|
||||
|
||||
Assert.assertEquals("思考中", fallback);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证思考事件会作为增量发送给前端。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldSendReasoningDeltaAsThinkingEnvelope() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.REASONING_DELTA);
|
||||
event.getPayload().put("reasoning", "思考增量");
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-1", emitter, new StringBuilder(), new ChatAssistantAccumulator(),
|
||||
chatContext(), new AtomicBoolean(false), false);
|
||||
|
||||
Assert.assertEquals(1, emitter.envelopes.size());
|
||||
Assert.assertEquals(ChatDomain.LLM, emitter.envelopes.get(0).getDomain());
|
||||
Assert.assertEquals(ChatType.THINKING, emitter.envelopes.get(0).getType());
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
|
||||
Assert.assertEquals("思考增量", payload.get("reasoning"));
|
||||
Assert.assertEquals("思考增量", payload.get("delta"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正文事件会作为增量发送给前端并累计到持久化缓冲。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldSendMessageDeltaAndAccumulateAnswer() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
StringBuilder answer = new StringBuilder();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.MESSAGE_DELTA);
|
||||
event.getPayload().put("text", "正文增量");
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-1", emitter, answer, new ChatAssistantAccumulator(),
|
||||
chatContext(), new AtomicBoolean(false), false);
|
||||
|
||||
Assert.assertEquals("正文增量", answer.toString());
|
||||
Assert.assertEquals(1, emitter.envelopes.size());
|
||||
Assert.assertEquals(ChatDomain.LLM, emitter.envelopes.get(0).getDomain());
|
||||
Assert.assertEquals(ChatType.MESSAGE, emitter.envelopes.get(0).getType());
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
|
||||
Assert.assertEquals("正文增量", payload.get("delta"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证自动上下文压缩事件会作为业务状态发送给前端。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldSendMemoryCompressionAsStatusEnvelope() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.MEMORY_COMPRESSION_STARTED);
|
||||
event.getPayload().put("statusKey", "memory-compression");
|
||||
event.getPayload().put("status", "running");
|
||||
event.getPayload().put("label", "正在整理上下文");
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-1", emitter, new StringBuilder(), new ChatAssistantAccumulator(),
|
||||
chatContext(), new AtomicBoolean(false), false);
|
||||
|
||||
Assert.assertEquals(1, emitter.envelopes.size());
|
||||
Assert.assertEquals(ChatDomain.BUSINESS, emitter.envelopes.get(0).getDomain());
|
||||
Assert.assertEquals(ChatType.STATUS, emitter.envelopes.get(0).getType());
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
|
||||
Assert.assertEquals("memory-compression", payload.get("statusKey"));
|
||||
Assert.assertEquals("running", payload.get("status"));
|
||||
Assert.assertEquals("正在整理上下文", payload.get("label"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证完成事件不会再次发送正文消息，只用于最终收口。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldNotSendMessageEnvelopeOnCompleted() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
setField(service, "agentRunRegistry", new AgentRunRegistry());
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
StringBuilder answer = new StringBuilder("流式正文");
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.COMPLETED);
|
||||
event.getPayload().put("text", "最终正文");
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-1", emitter, answer, new ChatAssistantAccumulator(),
|
||||
chatContext(), new AtomicBoolean(false), false);
|
||||
|
||||
Assert.assertEquals("最终正文", answer.toString());
|
||||
Assert.assertTrue(emitter.envelopes.stream().noneMatch(envelope ->
|
||||
envelope.getDomain() == ChatDomain.LLM && envelope.getType() == ChatType.MESSAGE));
|
||||
Assert.assertTrue(emitter.envelopes.stream().anyMatch(envelope ->
|
||||
envelope.getDomain() == ChatDomain.SYSTEM && envelope.getType() == ChatType.DONE));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证挂起事件后自然完成不会关闭 SSE 或清理运行态。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void finishIfNeededShouldKeepRuntimeWhenSuspended() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
setField(service, "agentRunRegistry", registry);
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
AtomicBoolean finished = new AtomicBoolean(false);
|
||||
AgentRunRegistry.AgentRunContext runContext = new AgentRunRegistry.AgentRunContext(
|
||||
"request-suspended",
|
||||
"session-suspended",
|
||||
new NoopRuntime(),
|
||||
emitter,
|
||||
chatContext(),
|
||||
new StringBuilder(),
|
||||
new ChatAssistantAccumulator(),
|
||||
finished,
|
||||
false,
|
||||
new AgentRunRegistry.RunOwner("agent-1", "session-suspended", "user-1"),
|
||||
null,
|
||||
event -> {
|
||||
},
|
||||
error -> {
|
||||
},
|
||||
() -> {
|
||||
}
|
||||
);
|
||||
registry.register(runContext);
|
||||
runContext.markSuspended();
|
||||
|
||||
invoke(service, "finishIfNeeded",
|
||||
new Class<?>[]{String.class, ChatSseEmitter.class, ChatRuntimeContext.class, StringBuilder.class,
|
||||
ChatAssistantAccumulator.class, AtomicBoolean.class, boolean.class},
|
||||
"request-suspended", emitter, chatContext(), new StringBuilder(),
|
||||
new ChatAssistantAccumulator(), finished, false);
|
||||
|
||||
Assert.assertFalse(finished.get());
|
||||
Assert.assertNotNull(registry.get("request-suspended"));
|
||||
Assert.assertTrue(emitter.envelopes.isEmpty());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证取消事件作为业务状态收口，不按系统错误发送。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldSendCancelledAsStatusAndDone() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
setField(service, "agentRunRegistry", new AgentRunRegistry());
|
||||
RecordingChatRuntimeManager chatRuntimeManager = new RecordingChatRuntimeManager();
|
||||
setField(service, "chatRuntimeManager", chatRuntimeManager);
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
StringBuilder answer = new StringBuilder("取消前正文");
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.CANCELLED);
|
||||
event.getPayload().put("reason", "用户拒绝执行");
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-1", emitter, answer, new ChatAssistantAccumulator(),
|
||||
chatContext(), new AtomicBoolean(false), true);
|
||||
|
||||
Assert.assertEquals(2, emitter.envelopes.size());
|
||||
Assert.assertEquals(ChatDomain.BUSINESS, emitter.envelopes.get(0).getDomain());
|
||||
Assert.assertEquals(ChatType.STATUS, emitter.envelopes.get(0).getType());
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
|
||||
Assert.assertEquals("agent-cancelled", payload.get("statusKey"));
|
||||
Assert.assertEquals("cancelled", payload.get("status"));
|
||||
Assert.assertEquals("用户拒绝执行", payload.get("message"));
|
||||
Assert.assertEquals(ChatDomain.SYSTEM, emitter.envelopes.get(1).getDomain());
|
||||
Assert.assertEquals(ChatType.DONE, emitter.envelopes.get(1).getType());
|
||||
Assert.assertEquals(1, chatRuntimeManager.recordAssistantCompletedCount);
|
||||
Assert.assertEquals("取消前正文", chatRuntimeManager.lastAssistantMessage.getContentText());
|
||||
Assert.assertEquals(1, chatRuntimeManager.recordFailureCount);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证最终知识库引用会保留命中分片原文。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void buildKnowledgeCitationPayloadShouldExposeChunkContent() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.COMPLETED);
|
||||
AgentMessage message = AgentMessage.text(AgentMessageRole.ASSISTANT, "answer");
|
||||
AgentKnowledgeReference reference = new AgentKnowledgeReference();
|
||||
reference.setKnowledgeId("kb-1");
|
||||
reference.setKnowledgeName("学生事务 FAQ");
|
||||
reference.setDocumentId("faq-1");
|
||||
reference.setDocumentName("faq-1.md");
|
||||
reference.setChunkId("chunk-1");
|
||||
reference.setChunkContent("暑假安排原文");
|
||||
reference.setScore(0.91D);
|
||||
reference.getMetadata().put("knowledgeType", "FAQ");
|
||||
reference.getMetadata().put("faqCollection", true);
|
||||
message.setKnowledgeReferences(List.of(reference));
|
||||
event.setMessage(message);
|
||||
|
||||
List<Map<String, Object>> payload = invoke(service, "buildKnowledgeCitationPayload",
|
||||
new Class<?>[]{AgentRuntimeEvent.class}, event);
|
||||
|
||||
Assert.assertEquals(1, payload.size());
|
||||
Assert.assertEquals("学生事务 FAQ", payload.get(0).get("knowledgeName"));
|
||||
Assert.assertEquals("暑假安排原文", payload.get(0).get("chunkContent"));
|
||||
Assert.assertEquals("FAQ", payload.get(0).get("knowledgeType"));
|
||||
Assert.assertEquals(Boolean.TRUE, payload.get(0).get("faqCollection"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证未保存草稿会生成临时 Agent ID，并把绑定指向该运行 ID。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void buildDraftAgentShouldGenerateRuntimeIdForUnsavedAgent() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentDraftChatRequest request = new AgentDraftChatRequest();
|
||||
Agent agent = new Agent();
|
||||
agent.setModelId(BigInteger.valueOf(10));
|
||||
request.setAgent(agent);
|
||||
|
||||
AgentToolBinding toolBinding = new AgentToolBinding();
|
||||
toolBinding.setToolType("PLUGIN");
|
||||
toolBinding.setTargetId(BigInteger.valueOf(20));
|
||||
toolBinding.setResourceSnapshot(Map.of("name", "client-forged"));
|
||||
toolBinding.setResourceSummary(Map.of("name", "client-forged"));
|
||||
request.setToolBindings(List.of(toolBinding));
|
||||
|
||||
AgentKnowledgeBinding knowledgeBinding = new AgentKnowledgeBinding();
|
||||
knowledgeBinding.setKnowledgeId(BigInteger.valueOf(30));
|
||||
knowledgeBinding.setResourceSnapshot(Map.of("title", "client-forged"));
|
||||
knowledgeBinding.setResourceSummary(Map.of("title", "client-forged"));
|
||||
request.setKnowledgeBindings(List.of(knowledgeBinding));
|
||||
|
||||
LoginAccount account = new LoginAccount();
|
||||
account.setId(BigInteger.ONE);
|
||||
account.setTenantId(BigInteger.valueOf(2));
|
||||
account.setDeptId(BigInteger.valueOf(3));
|
||||
|
||||
Agent draftAgent = invoke(service, "buildDraftAgent",
|
||||
new Class<?>[]{AgentDraftChatRequest.class, LoginAccount.class}, request, account);
|
||||
|
||||
Assert.assertNotNull(draftAgent.getId());
|
||||
Assert.assertEquals(BigInteger.valueOf(2), draftAgent.getTenantId());
|
||||
Assert.assertEquals(draftAgent.getId(), draftAgent.getToolBindings().get(0).getAgentId());
|
||||
Assert.assertEquals(draftAgent.getId(), draftAgent.getKnowledgeBindings().get(0).getAgentId());
|
||||
Assert.assertTrue(draftAgent.getToolBindings().get(0).getEnabled());
|
||||
Assert.assertTrue(draftAgent.getKnowledgeBindings().get(0).getEnabled());
|
||||
Assert.assertTrue(draftAgent.getToolBindings().get(0).getResourceSnapshot().isEmpty());
|
||||
Assert.assertTrue(draftAgent.getToolBindings().get(0).getResourceSummary().isEmpty());
|
||||
Assert.assertTrue(draftAgent.getKnowledgeBindings().get(0).getResourceSnapshot().isEmpty());
|
||||
Assert.assertTrue(draftAgent.getKnowledgeBindings().get(0).getResourceSummary().isEmpty());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正式聊天在获取运行锁失败时不会提前写入用户消息。
|
||||
*
|
||||
* <p>运行锁是 AgentScope session 与 chatlog 的一致性入口。若同会话并发请求抢锁失败，
|
||||
* 必须在 prepareSession 和 recordUserMessage 之前失败，避免 chatlog 出现没有真实运行的用户消息。</p>
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void runShouldAcquireLockBeforePersistingUserMessage() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
RecordingChatRuntimeManager chatRuntimeManager = new RecordingChatRuntimeManager();
|
||||
setField(service, "chatRuntimeManager", chatRuntimeManager);
|
||||
setField(service, "agentRunLock", new RejectingAgentRunLock());
|
||||
|
||||
Agent agent = new Agent();
|
||||
agent.setId(BigInteger.valueOf(100));
|
||||
ChatRuntimeContext context = chatContext();
|
||||
|
||||
Exception thrown = Assert.assertThrows(Exception.class, () -> invoke(service, "run",
|
||||
new Class<?>[]{Agent.class, String.class, String.class, String.class, String.class,
|
||||
String.class, ChatRuntimeContext.class, boolean.class},
|
||||
agent, "你好", "request-lock", "trace-lock", "session-lock", "AGENT", context, true));
|
||||
|
||||
Assert.assertTrue(rootCause(thrown) instanceof BusinessException);
|
||||
Assert.assertEquals(0, chatRuntimeManager.prepareSessionCount);
|
||||
Assert.assertEquals(0, chatRuntimeManager.recordUserMessageCount);
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正式聊天会在会话准备完成后向前端返回真实会话 ID。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void sendSessionCreatedShouldExposeSessionId() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
RecordingChatSseEmitter emitter = new RecordingChatSseEmitter();
|
||||
|
||||
Boolean sent = invoke(service, "sendSessionCreated",
|
||||
new Class<?>[]{ChatSseEmitter.class, BigInteger.class}, emitter, BigInteger.valueOf(123));
|
||||
|
||||
Assert.assertTrue(sent);
|
||||
Assert.assertEquals(1, emitter.envelopes.size());
|
||||
Assert.assertEquals(ChatDomain.SYSTEM, emitter.envelopes.get(0).getDomain());
|
||||
Assert.assertEquals(ChatType.SESSION_CREATED, emitter.envelopes.get(0).getType());
|
||||
@SuppressWarnings("unchecked")
|
||||
Map<String, Object> payload = (Map<String, Object>) emitter.envelopes.get(0).getPayload();
|
||||
Assert.assertEquals("123", payload.get("sessionId"));
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正式聊天只有新会话首轮会自动设置默认标题。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void applyFormalSessionTitleShouldOnlyUseFirstPromptForNewSession() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
ChatRuntimeContext newContext = chatContext();
|
||||
ChatRuntimeContext existingContext = chatContext();
|
||||
ChatSessionSummary existingSession = new ChatSessionSummary();
|
||||
existingSession.setId(BigInteger.valueOf(123));
|
||||
existingSession.setTitle("用户改过的标题");
|
||||
existingSession.setMessageCount(2);
|
||||
|
||||
invoke(service, "applyFormalSessionTitle",
|
||||
new Class<?>[]{ChatRuntimeContext.class, String.class, ChatSessionSummary.class},
|
||||
newContext, "第一句话", null);
|
||||
invoke(service, "applyFormalSessionTitle",
|
||||
new Class<?>[]{ChatRuntimeContext.class, String.class, ChatSessionSummary.class},
|
||||
existingContext, "后续消息", existingSession);
|
||||
|
||||
Assert.assertEquals("第一句话", newContext.getSessionTitle());
|
||||
Assert.assertNull(existingContext.getSessionTitle());
|
||||
}
|
||||
|
||||
/**
|
||||
* 验证正式聊天 SSE 断开后会取消运行并保存断开前已输出的 assistant 内容。
|
||||
*
|
||||
* @throws Exception 反射调用失败时抛出
|
||||
*/
|
||||
@Test
|
||||
public void handleRuntimeEventShouldCancelAndRecordFailureWhenSseDisconnected() throws Exception {
|
||||
AgentRunService service = new AgentRunService();
|
||||
AgentRunRegistry registry = new AgentRunRegistry();
|
||||
RecordingChatRuntimeManager chatRuntimeManager = new RecordingChatRuntimeManager();
|
||||
setField(service, "agentRunRegistry", registry);
|
||||
setField(service, "chatRuntimeManager", chatRuntimeManager);
|
||||
|
||||
AtomicBoolean finished = new AtomicBoolean(false);
|
||||
ChatRuntimeContext context = chatContext();
|
||||
AgentRunRegistry.AgentRunContext runContext = new AgentRunRegistry.AgentRunContext(
|
||||
"request-disconnected",
|
||||
"session-disconnected",
|
||||
new NoopRuntime(),
|
||||
new FailingChatSseEmitter(),
|
||||
context,
|
||||
new StringBuilder(),
|
||||
new ChatAssistantAccumulator(),
|
||||
finished,
|
||||
true,
|
||||
new AgentRunRegistry.RunOwner("agent-1", "session-disconnected", "user-1"),
|
||||
null,
|
||||
ignored -> {
|
||||
},
|
||||
ignored -> {
|
||||
},
|
||||
() -> {
|
||||
}
|
||||
);
|
||||
registry.register(runContext);
|
||||
AgentRuntimeEvent event = AgentRuntimeEvent.of(AgentRuntimeEventType.MESSAGE_DELTA);
|
||||
event.getPayload().put("text", "断连前正文");
|
||||
StringBuilder answer = new StringBuilder();
|
||||
ChatAssistantAccumulator assistantAccumulator = new ChatAssistantAccumulator();
|
||||
|
||||
invoke(service, "handleRuntimeEvent",
|
||||
runtimeEventParameterTypes(),
|
||||
event, "request-disconnected", new FailingChatSseEmitter(), answer,
|
||||
assistantAccumulator, context, finished, true);
|
||||
|
||||
Assert.assertTrue(finished.get());
|
||||
Assert.assertNull(registry.get("request-disconnected"));
|
||||
Assert.assertEquals(1, chatRuntimeManager.recordAssistantCompletedCount);
|
||||
Assert.assertEquals("断连前正文", chatRuntimeManager.lastAssistantMessage.getContentText());
|
||||
Assert.assertEquals(1, chatRuntimeManager.recordFailureCount);
|
||||
Assert.assertEquals(0, chatRuntimeManager.recordCompletedCount);
|
||||
}
|
||||
|
||||
@SuppressWarnings("unchecked")
|
||||
private <T> T invoke(Object target, String methodName, Class<?>[] parameterTypes, Object... args) throws Exception {
|
||||
Method method = target.getClass().getDeclaredMethod(methodName, parameterTypes);
|
||||
method.setAccessible(true);
|
||||
return (T) method.invoke(target, args);
|
||||
}
|
||||
|
||||
private void setField(Object target, String fieldName, Object value) throws Exception {
|
||||
java.lang.reflect.Field field = target.getClass().getDeclaredField(fieldName);
|
||||
field.setAccessible(true);
|
||||
field.set(target, value);
|
||||
}
|
||||
|
||||
private Class<?>[] runtimeEventParameterTypes() {
|
||||
return new Class<?>[]{AgentRuntimeEvent.class, String.class, ChatSseEmitter.class, StringBuilder.class,
|
||||
ChatAssistantAccumulator.class,
|
||||
ChatRuntimeContext.class, AtomicBoolean.class, boolean.class};
|
||||
}
|
||||
|
||||
private ChatRuntimeContext chatContext() {
|
||||
ChatRuntimeContext context = new ChatRuntimeContext();
|
||||
context.setAssistantId(BigInteger.valueOf(100));
|
||||
context.setAssistantName("Agent");
|
||||
context.setUserId(BigInteger.valueOf(101));
|
||||
context.setUserName("用户");
|
||||
return context;
|
||||
}
|
||||
|
||||
private Throwable rootCause(Throwable throwable) {
|
||||
Throwable current = throwable;
|
||||
while (current.getCause() != null) {
|
||||
current = current.getCause();
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
/**
|
||||
* 记录发送内容的 SSE 测试桩。
|
||||
*/
|
||||
private static class RecordingChatSseEmitter extends ChatSseEmitter {
|
||||
|
||||
private final List<ChatEnvelope<?>> envelopes = new java.util.ArrayList<>();
|
||||
|
||||
@Override
|
||||
public boolean send(ChatEnvelope<?> envelope) {
|
||||
envelopes.add(envelope);
|
||||
return true;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean sendDone(ChatEnvelope<?> envelope) {
|
||||
envelopes.add(envelope);
|
||||
return true;
|
||||
}
|
||||
}
|
||||
|
||||
private static class FailingChatSseEmitter extends ChatSseEmitter {
|
||||
|
||||
@Override
|
||||
public boolean send(ChatEnvelope<?> envelope) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean sendDone(ChatEnvelope<?> envelope) {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
private static class NoopRuntime implements com.easyagents.agent.runtime.AgentRuntime {
|
||||
|
||||
@Override
|
||||
public void init(com.easyagents.agent.runtime.AgentInitRequest request) {
|
||||
// 测试桩无需初始化。
|
||||
}
|
||||
|
||||
@Override
|
||||
public reactor.core.publisher.Flux<AgentRuntimeEvent> stream(AgentMessage userMessage) {
|
||||
return reactor.core.publisher.Flux.empty();
|
||||
}
|
||||
|
||||
@Override
|
||||
public reactor.core.publisher.Flux<AgentRuntimeEvent> resume(com.easyagents.agent.runtime.AgentResumeRequest request) {
|
||||
return reactor.core.publisher.Flux.empty();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 记录 chatlog 写入动作的测试桩。
|
||||
*/
|
||||
private static class RecordingChatRuntimeManager implements ChatRuntimeManager {
|
||||
|
||||
private int prepareSessionCount;
|
||||
private int recordUserMessageCount;
|
||||
private int recordAssistantCompletedCount;
|
||||
private int recordFailureCount;
|
||||
private int recordCompletedCount;
|
||||
private ChatRuntimeMessage lastAssistantMessage;
|
||||
|
||||
@Override
|
||||
public void prepareSession(ChatRuntimeContext context) {
|
||||
prepareSessionCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void recordUserMessage(ChatRuntimeContext context, ChatRuntimeMessage message) {
|
||||
recordUserMessageCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void recordAssistantDelta(ChatRuntimeContext context, ChatRuntimeMessage message) {
|
||||
// 测试桩无需记录。
|
||||
}
|
||||
|
||||
@Override
|
||||
public void recordAssistantCompleted(ChatRuntimeContext context, ChatRuntimeMessage message) {
|
||||
recordAssistantCompletedCount++;
|
||||
lastAssistantMessage = message;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void recordFailure(ChatRuntimeContext context, Throwable throwable) {
|
||||
recordFailureCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void recordCompleted(ChatRuntimeContext context) {
|
||||
recordCompletedCount++;
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<ChatRuntimeMessage> loadMessages(ChatRuntimeContext context, int limit) {
|
||||
return List.of();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 模拟同会话运行锁已被占用的测试桩。
|
||||
*/
|
||||
private static class RejectingAgentRunLock implements AgentRunLock {
|
||||
|
||||
@Override
|
||||
public Handle acquire(BigInteger agentId, String sessionId) {
|
||||
throw new BusinessException("当前 Agent 会话已有运行中的请求，请稍后再试");
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user