Skip to content

P
Pangea-Agent
Commit 12089728 authored by youxiaoji's avatar youxiaoji
Browse files
Options

* [区分智能体，非访客预约智能体暂不添加toolcontext]

parent 3f92e64d
Hide whitespace changes
Inline Side-by-side
Showing with 100 additions and 86 deletions
backend/src/main/java/pangea/hiagent/agent/react/DefaultReactExecutor.java
View file @ 12089728
......@@ -34,74 +34,75 @@ public class DefaultReactExecutor implements ReactExecutor {
private final UserSseService userSseService;
@Value("${hiagent.react.system-prompt}")
private String defaultSystemPrompt;
private final List<ReactCallback> reactCallbacks = new ArrayList<>();
private final EventSplitter eventSplitter;
private MemoryService memoryService;
private ErrorHandlerService errorHandlerService;
private final AgentToolManager agentToolManager;
private final UserTokenService userTokenService;
public DefaultReactExecutor(EventSplitter eventSplitter, AgentToolManager agentToolManager ,
public DefaultReactExecutor(EventSplitter eventSplitter, AgentToolManager agentToolManager,
MemoryService memoryService, ErrorHandlerService errorHandlerService, UserSseService userSseService, UserTokenService userTokenService) {
this.eventSplitter = eventSplitter;
this.agentToolManager = agentToolManager;
this.memoryService = memoryService;
this.memoryService = memoryService;
this.errorHandlerService = errorHandlerService;
this.userSseService = userSseService;
this.userTokenService = userTokenService;
}
@Override
public void addReactCallback(ReactCallback callback) {
if (callback != null) {
reactCallbacks.add(callback);
}
}
@Override
public String execute(ChatClient chatClient, String userInput, List<Object> tools, Agent agent) {
// 调用带用户ID的方法，首先尝试获取当前用户ID
String userId = UserUtils.getCurrentUserIdStatic();
return execute(chatClient, userInput, tools, agent, userId);
}
@Override
public String execute(ChatClient chatClient, String userInput, List<Object> tools, Agent agent, String userId) {
log.info("开始执行ReAct流程，用户输入: {}", userInput);
List<Object> agentTools = getAgentTools(agent);
try {
Prompt prompt = buildPromptWithHistory(defaultSystemPrompt, userInput, agent, userId);
ChatResponse response = chatClient.prompt(prompt)
.tools(agentTools.toArray())
.call()
.chatResponse();
String responseText = response.getResult().getOutput().getText();
log.info("最终答案: {}", responseText);
// 保存助手回复到内存，使用提供的用户ID
saveAssistantResponseToMemory(agent, responseText, userId);
return responseText;
} catch (Exception e) {
log.error("执行ReAct流程时发生错误", e);
return handleReActError(e);
}
}
/**
* 处理ReAct执行过程中发生的错误
*
*
* @param e 发生的异常
* @return 错误处理结果
*/
......@@ -109,21 +110,21 @@ public class DefaultReactExecutor implements ReactExecutor {
log.error("ReAct执行过程中发生错误", e);
return errorHandlerService.handleSyncError(e, "处理ReAct请求时发生错误");
}
/**
* 构建带有历史记录的提示词
*
*
* @param systemPrompt 系统提示词
* @param userInput 用户输入
* @param agent 智能体对象
* @param userId 用户ID（可选，如果为null则自动获取）
* @param userInput 用户输入
* @param agent 智能体对象
* @param userId 用户ID（可选，如果为null则自动获取）
* @return 构建好的提示词对象
*/
private Prompt buildPromptWithHistory(String systemPrompt, String userInput, Agent agent, String userId) {
List<org.springframework.ai.chat.messages.Message> messages = new ArrayList<>();
messages.add(new SystemMessage(systemPrompt));
if (agent != null) {
try {
// 如果没有提供用户ID，则尝试获取当前用户ID
......@@ -131,80 +132,93 @@ public class DefaultReactExecutor implements ReactExecutor {
userId = UserUtils.getCurrentUserIdStatic();
}
String sessionId = memoryService.generateSessionId(agent, userId);
int historyLength = agent.getHistoryLength() != null ? agent.getHistoryLength() : 10;
List<org.springframework.ai.chat.messages.Message> historyMessages =
memoryService.getHistoryMessages(sessionId, historyLength);
List<org.springframework.ai.chat.messages.Message> historyMessages =
memoryService.getHistoryMessages(sessionId, historyLength);
messages.addAll(historyMessages);
memoryService.addUserMessageToMemory(sessionId, userInput);
} catch (Exception e) {
log.warn("获取历史对话记录时发生错误: {}", e.getMessage());
}
}
messages.add(new UserMessage(userInput));
return new Prompt(messages);
}
@Override
public void executeStream(ChatClient chatClient, String userInput, List<Object> tools, Consumer<String> tokenConsumer, Agent agent) {
// 调用带用户ID的方法，但首先尝试获取当前用户ID
String userId = UserUtils.getCurrentUserIdStatic();
executeStream(chatClient, userInput, tools, tokenConsumer, agent, userId);
}
@Override
public void executeStream(ChatClient chatClient, String userInput, List<Object> tools, Consumer<String> tokenConsumer, Agent agent, String userId) {
log.info("使用stream()方法处理ReAct流程，支持真正的流式输出");
List<Object> agentTools = getAgentTools(agent);
StringBuilder fullResponse = new StringBuilder();
try {
Prompt prompt = buildPromptWithHistory(defaultSystemPrompt, userInput, agent, userId);
SseTokenEmitter sseTokenEmitter = (SseTokenEmitter)tokenConsumer;
SseTokenEmitter sseTokenEmitter = (SseTokenEmitter) tokenConsumer;
String emitterId = sseTokenEmitter.getEmitterId();
chatClient.prompt(prompt)
.tools(agentTools.toArray())
.toolContext(Map.of("emitterId", emitterId, "userId", sseTokenEmitter.getUserId()))
.stream()
.chatResponse()
.subscribe(
chatResponse -> handleTokenResponse(chatResponse, tokenConsumer, fullResponse),
throwable -> handleStreamError(throwable, tokenConsumer,emitterId),
() -> handleStreamCompletion(tokenConsumer, fullResponse, agent, userId,emitterId)
);
} catch (Exception e) {
if (agent.getId().contains("agent-8")) {
chatClient.prompt(prompt)
.tools(agentTools.toArray())
.toolContext(Map.of("emitterId", emitterId, "userId", sseTokenEmitter.getUserId()))
.stream()
.chatResponse()
.subscribe(
chatResponse -> handleTokenResponse(chatResponse, tokenConsumer, fullResponse),
throwable -> handleStreamError(throwable, tokenConsumer, emitterId),
() -> handleStreamCompletion(tokenConsumer, fullResponse, agent, userId, emitterId)
);
} else {
chatClient.prompt(prompt)
.tools(agentTools.toArray())
.stream()
.chatResponse()
.subscribe(
chatResponse -> handleTokenResponse(chatResponse, tokenConsumer, fullResponse),
throwable -> handleStreamError(throwable, tokenConsumer, emitterId),
() -> handleStreamCompletion(tokenConsumer, fullResponse, agent, userId, emitterId)
);
}
} catch (Exception e) {
log.error("流式执行ReAct流程时发生错误", e);
errorHandlerService.handleReactFlowError(e, tokenConsumer);
}
}
/**
* 处理流式响应中的单个token
*
* @param chatResponse 聊天响应对象
*
* @param chatResponse 聊天响应对象
* @param tokenConsumer token消费者
* @param fullResponse 完整响应构建器
* @param fullResponse 完整响应构建器
*/
private void handleTokenResponse(org.springframework.ai.chat.model.ChatResponse chatResponse, Consumer<String> tokenConsumer, StringBuilder fullResponse) {
try {
String token = chatResponse.getResult().getOutput().getText();
if (isValidToken(token)) {
fullResponse.append(token);
if (tokenConsumer != null) {
tokenConsumer.accept(token);
}
eventSplitter.feedToken(token);
}
} catch (Exception e) {
......@@ -212,22 +226,22 @@ public class DefaultReactExecutor implements ReactExecutor {
errorHandlerService.handleReactFlowError(e, tokenConsumer);
}
}
/**
* 处理流式响应完成事件
*
*
* @param tokenConsumer token消费者
* @param fullResponse 完整响应内容
* @param agent 智能体对象
* @param userId 用户ID
* @param fullResponse 完整响应内容
* @param agent 智能体对象
* @param userId 用户ID
*/
private void handleStreamCompletion(Consumer<String> tokenConsumer, StringBuilder fullResponse, Agent agent, String userId, String emitterId) {
try {
log.info("流式处理完成");
String responseStr = fullResponse.toString();
saveAssistantResponseToMemory(agent, responseStr, userId);
log.info("complete, remove emitterId {}",emitterId);
log.info("complete, remove emitterId {}", emitterId);
userSseService.removeEmitter(emitterId);
sendCompletionEvent(tokenConsumer, responseStr);
} catch (Exception e) {
......@@ -235,13 +249,13 @@ public class DefaultReactExecutor implements ReactExecutor {
handleCompletionError(tokenConsumer, e);
}
}
/**
* 将助手的回复保存到内存中
*
* @param agent 智能体对象
*
* @param agent 智能体对象
* @param response 助手的回复内容
* @param userId 用户ID
* @param userId 用户ID
*/
private void saveAssistantResponseToMemory(Agent agent, String response, String userId) {
if (agent != null) {
......@@ -253,12 +267,12 @@ public class DefaultReactExecutor implements ReactExecutor {
}
}
}
/**
* 处理完成事件时发生的错误
*
*
* @param tokenConsumer token消费者
* @param e 发生的异常
* @param e 发生的异常
*/
private void handleCompletionError(Consumer<String> tokenConsumer, Exception e) {
if (tokenConsumer instanceof TokenConsumerWithCompletion) {
......@@ -273,40 +287,40 @@ public class DefaultReactExecutor implements ReactExecutor {
}
}
}
/**
* 验证token是否有效
*
*
* @param token 待验证的token
* @return 如果token有效则返回true，否则返回false
*/
private boolean isValidToken(String token) {
return token != null && !token.isEmpty();
}
/**
* 处理流式响应中的错误
*
* @param throwable 异常对象
*
* @param throwable 异常对象
* @param tokenConsumer token消费者
*/
private void handleStreamError(Throwable throwable, Consumer<String> tokenConsumer,String emitterId) {
private void handleStreamError(Throwable throwable, Consumer<String> tokenConsumer, String emitterId) {
log.info("error,remove emitterId:{}", emitterId);
userSseService.removeEmitter(emitterId);
errorHandlerService.handleStreamError(throwable, tokenConsumer, "ReAct流式处理");
}
/**
* 发送完成事件
*
*
* @param tokenConsumer token消费者
* @param fullResponse 完整响应内容
* @param fullResponse 完整响应内容
*/
private void sendCompletionEvent(Consumer<String> tokenConsumer, String fullResponse) {
if (fullResponse == null) {
fullResponse = "";
}
if (tokenConsumer instanceof TokenConsumerWithCompletion) {
try {
((TokenConsumerWithCompletion) tokenConsumer).onComplete(fullResponse);
......@@ -327,16 +341,16 @@ public class DefaultReactExecutor implements ReactExecutor {
tokenConsumer.accept("");
}
}
/**
* 获取智能体工具
*
*
* @param agent 智能体对象
* @return 智能体可用的工具列表
*/
private List<Object> getAgentTools(Agent agent) {
List<Object> tools = new ArrayList<>();
if (agent != null) {
try {
tools = agentToolManager.getAvailableToolInstances(agent);
......@@ -345,7 +359,7 @@ public class DefaultReactExecutor implements ReactExecutor {
// 发生异常时，tools 保持为空列表
}
}
return tools;
}
}
\ No newline at end of file
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment