mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-09-13 19:23:42 +08:00
720 lines
33 KiB
Java
720 lines
33 KiB
Java
package vip.mate.agent;
|
||
|
||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||
import lombok.RequiredArgsConstructor;
|
||
import lombok.extern.slf4j.Slf4j;
|
||
import org.springframework.beans.factory.annotation.Autowired;
|
||
import org.springframework.context.ApplicationEventPublisher;
|
||
import org.springframework.context.event.EventListener;
|
||
import org.springframework.stereotype.Service;
|
||
import org.springframework.util.StringUtils;
|
||
import reactor.core.publisher.Flux;
|
||
import vip.mate.agent.context.ChatOrigin;
|
||
import vip.mate.agent.context.ChatOriginHolder;
|
||
import vip.mate.agent.event.AgentLifecycleEvent;
|
||
import vip.mate.agent.model.AgentEntity;
|
||
import vip.mate.agent.repository.AgentMapper;
|
||
import vip.mate.exception.MateClawException;
|
||
import vip.mate.llm.chatmodel.ThinkingLevelHolder;
|
||
import vip.mate.llm.event.ModelConfigChangedEvent;
|
||
import vip.mate.memory.MemoryProperties;
|
||
import vip.mate.memory.lifecycle.MemoryLifecycleMediator;
|
||
import vip.mate.memory.lifecycle.TurnContext;
|
||
import vip.mate.memory.service.MemoryRecallTracker;
|
||
import vip.mate.workspace.conversation.model.ConversationEntity;
|
||
import vip.mate.workspace.conversation.repository.ConversationMapper;
|
||
|
||
import java.util.List;
|
||
import java.time.Duration;
|
||
import java.util.Map;
|
||
import java.util.concurrent.ConcurrentHashMap;
|
||
import java.util.function.Function;
|
||
import java.util.function.Supplier;
|
||
|
||
/**
|
||
* Agent 业务服务
|
||
* <p>
|
||
* 负责 Agent 的 CRUD 管理和运行时实例管理。
|
||
* 构建逻辑委托给 {@link AgentGraphBuilder}。
|
||
*
|
||
* @author MateClaw Team
|
||
*/
|
||
@Slf4j
|
||
@Service
|
||
@RequiredArgsConstructor
|
||
public class AgentService {
|
||
|
||
private final AgentMapper agentMapper;
|
||
private final AgentGraphBuilder agentGraphBuilder;
|
||
private final MemoryRecallTracker memoryRecallTracker;
|
||
private final MemoryLifecycleMediator lifecycleMediator;
|
||
private final MemoryProperties memoryProperties;
|
||
private final vip.mate.memory.identity.MemoryOwnerResolver memoryOwnerResolver;
|
||
/** Read-only lookup of a conversation's pinned model. Mapper (not service)
|
||
* to keep this a leaf dependency with no risk of a bean cycle. */
|
||
private final ConversationMapper conversationMapper;
|
||
|
||
/** Field-injected publisher for agent_lifecycle trigger events; the
|
||
* trigger module's bridge listens and forwards into ingest. */
|
||
@Autowired(required = false)
|
||
private ApplicationEventPublisher events;
|
||
|
||
/**
|
||
* Runtime Agent instance cache. Keyed first by agentId, then by a model
|
||
* key, so a conversation that pins a non-default model gets its own graph
|
||
* variant instead of mutating the one every other conversation shares.
|
||
* The model key is {@code ""} for the Agent / global-default model.
|
||
*/
|
||
private final Map<Long, Map<String, BaseAgent>> agentInstances = new ConcurrentHashMap<>();
|
||
|
||
// ==================== CRUD ====================
|
||
|
||
public List<AgentEntity> listAgents() {
|
||
return agentMapper.selectList(new LambdaQueryWrapper<AgentEntity>()
|
||
.orderByDesc(AgentEntity::getCreateTime));
|
||
}
|
||
|
||
/**
|
||
* 按工作区列出 Agent
|
||
*/
|
||
public List<AgentEntity> listAgentsByWorkspace(Long workspaceId) {
|
||
return listAgentsByWorkspace(workspaceId, null);
|
||
}
|
||
|
||
/**
|
||
* 按工作区列出 Agent，可选过滤启用状态。
|
||
*
|
||
* @param enabled non-null restricts the result set to agents whose
|
||
* {@code enabled} column matches the given value.
|
||
* Pass {@code true} from chat selectors so disabled
|
||
* agents disappear from the picker; the admin
|
||
* management page passes {@code null} to keep
|
||
* disabled rows visible for re-enabling.
|
||
*/
|
||
public List<AgentEntity> listAgentsByWorkspace(Long workspaceId, Boolean enabled) {
|
||
LambdaQueryWrapper<AgentEntity> q = new LambdaQueryWrapper<AgentEntity>()
|
||
.eq(AgentEntity::getWorkspaceId, workspaceId);
|
||
if (enabled != null) {
|
||
q.eq(AgentEntity::getEnabled, enabled);
|
||
}
|
||
return agentMapper.selectList(q.orderByDesc(AgentEntity::getCreateTime));
|
||
}
|
||
|
||
public AgentEntity getAgent(Long id) {
|
||
AgentEntity entity = agentMapper.selectById(id);
|
||
if (entity == null) {
|
||
throw new MateClawException("err.agent.not_found", "Agent不存在: " + id);
|
||
}
|
||
return entity;
|
||
}
|
||
|
||
public AgentEntity createAgent(AgentEntity agent) {
|
||
agent.setEnabled(true);
|
||
if (agent.getAgentType() == null) {
|
||
agent.setAgentType("react");
|
||
}
|
||
requireUniqueName(agent, null);
|
||
agentMapper.insert(agent);
|
||
publishLifecycle(agent, "spawned");
|
||
return agent;
|
||
}
|
||
|
||
public AgentEntity updateAgent(AgentEntity agent) {
|
||
// Detect enabled-flag flip so the lifecycle event reflects the
|
||
// intent rather than every metadata edit. Reading the prior row
|
||
// is cheap and gives us a clean diff source.
|
||
AgentEntity prior = agentMapper.selectById(agent.getId());
|
||
// Only re-validate uniqueness when the name actually changes —
|
||
// a pure metadata edit (icon, prompt, ...) shouldn't pay the
|
||
// SELECT cost or risk a false positive against the row itself.
|
||
if (prior != null
|
||
&& agent.getName() != null
|
||
&& !agent.getName().equals(prior.getName())) {
|
||
// Workspace cannot be moved (Controller pins it to prior.workspaceId),
|
||
// so reuse it for the lookup even if the incoming DTO left it null.
|
||
if (agent.getWorkspaceId() == null) {
|
||
agent.setWorkspaceId(prior.getWorkspaceId());
|
||
}
|
||
requireUniqueName(agent, agent.getId());
|
||
}
|
||
agentMapper.updateById(agent);
|
||
agentInstances.remove(agent.getId());
|
||
if (prior != null && prior.getEnabled() != null
|
||
&& !prior.getEnabled().equals(agent.getEnabled())) {
|
||
publishLifecycle(agent,
|
||
Boolean.TRUE.equals(agent.getEnabled()) ? "enabled" : "disabled");
|
||
}
|
||
return agent;
|
||
}
|
||
|
||
/**
|
||
* Friendly business-code surface for the {@code (workspace_id, name)}
|
||
* unique index added in V102.
|
||
*
|
||
* <p>The wire shape is the project-wide R<T> envelope: HTTP status
|
||
* stays 200 (per the convention in {@code R.fail} and the axios
|
||
* interceptor in {@code mateclaw-ui/src/api/index.ts}); the 409 lives in
|
||
* the response body's {@code code} field so the front-end can branch
|
||
* without breaking on an axios error. Without this pre-check the
|
||
* duplicate save would surface as an opaque
|
||
* {@code DataIntegrityViolation} stack trace.
|
||
*
|
||
* @param excludeId when non-null, skip this row in the lookup so
|
||
* {@link #updateAgent} doesn't mistake the row for its
|
||
* own duplicate.
|
||
*/
|
||
private void requireUniqueName(AgentEntity agent, Long excludeId) {
|
||
if (agent.getName() == null || agent.getName().isBlank()) {
|
||
throw new MateClawException("err.agent.name_required", 400, "Agent 名称不能为空");
|
||
}
|
||
Long workspaceId = agent.getWorkspaceId() == null ? 1L : agent.getWorkspaceId();
|
||
LambdaQueryWrapper<AgentEntity> q = new LambdaQueryWrapper<AgentEntity>()
|
||
.eq(AgentEntity::getWorkspaceId, workspaceId)
|
||
.eq(AgentEntity::getName, agent.getName());
|
||
if (excludeId != null) {
|
||
q.ne(AgentEntity::getId, excludeId);
|
||
}
|
||
Long count = agentMapper.selectCount(q);
|
||
if (count != null && count > 0) {
|
||
throw new MateClawException("err.agent.duplicate_name", 409,
|
||
"工作区内已存在同名 Agent: " + agent.getName());
|
||
}
|
||
}
|
||
|
||
public void deleteAgent(Long id) {
|
||
AgentEntity prior = agentMapper.selectById(id);
|
||
agentMapper.deleteById(id);
|
||
agentInstances.remove(id);
|
||
if (prior != null) publishLifecycle(prior, "terminated");
|
||
}
|
||
|
||
/**
|
||
* Best-effort publish of an {@link AgentLifecycleEvent}. A publish
|
||
* failure must never roll back the agent CRUD that just succeeded —
|
||
* the agent_lifecycle trigger surface is observability, not the
|
||
* canonical record.
|
||
*/
|
||
private void publishLifecycle(AgentEntity agent, String phase) {
|
||
if (events == null || agent == null) return;
|
||
try {
|
||
events.publishEvent(new AgentLifecycleEvent(
|
||
agent.getWorkspaceId() == null ? 0L : agent.getWorkspaceId(),
|
||
agent.getId() == null ? 0L : agent.getId(),
|
||
agent.getName(),
|
||
phase,
|
||
System.currentTimeMillis()));
|
||
} catch (Exception e) {
|
||
log.warn("[AgentService] lifecycle publish failed for agent {} ({}): {}",
|
||
agent.getId(), phase, e.getMessage());
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 清除 Agent 运行时缓存（绑定变更后需调用，使下次对话重新构建 Agent）
|
||
*/
|
||
public void invalidateAgentCache(Long agentId) {
|
||
agentInstances.remove(agentId);
|
||
}
|
||
|
||
/**
|
||
* Invalidate the cached agent instance whenever one of its workspace files
|
||
* changes. The system prompt (which embeds MEMORY.md / PROFILE.md / structured
|
||
* memory) is baked into the cached instance at build time, so memory edits made
|
||
* via tools, consolidation, or cleanup would otherwise stay invisible until an
|
||
* agent config change or restart. Rebuilding on the next turn picks them up.
|
||
*/
|
||
@org.springframework.context.event.EventListener
|
||
public void onWorkspaceFileChanged(vip.mate.workspace.document.event.WorkspaceFileChangedEvent event) {
|
||
if (event.agentId() != null) {
|
||
agentInstances.remove(event.agentId());
|
||
}
|
||
}
|
||
|
||
// ==================== 运行时入口 ====================
|
||
|
||
public String chat(Long agentId, String message, String conversationId) {
|
||
return chat(agentId, message, conversationId, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
/**
|
||
* RFC-063r §2.5: preferred entry — accepts the originating
|
||
* {@link ChatOrigin} so channel binding and workspace context propagate
|
||
* down to {@code @Tool} methods via Spring AI {@link org.springframework.ai.chat.model.ToolContext}.
|
||
*/
|
||
public String chat(Long agentId, String message, String conversationId, ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, message);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY);
|
||
try {
|
||
return withLifecycleSync(agentId, message, conversationId,
|
||
(msg, convId) -> agent.chat(msg, convId));
|
||
} finally {
|
||
ChatOriginHolder.clear();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Sync chat that also captures token usage and runtime model attribution
|
||
* from the agent graph's {@code _usage_final} event. Equivalent to
|
||
* subscribing to {@link #chatStructuredStream} and joining all content
|
||
* deltas — produces the same assistant text as {@link #chat} but exposes
|
||
* the usage figures so callers can persist them on the assistant message.
|
||
*
|
||
* <p>Prefer this entry over {@link #chat} for any path that writes the
|
||
* reply to {@code mate_message} (sync HTTP endpoint, voice WebSocket,
|
||
* cron task, post-approval replay); the plain {@link #chat} stays as the
|
||
* thin wrapper for fire-and-forget invocations where usage is not needed.
|
||
*/
|
||
public ChatResult chatWithUsage(Long agentId, String message, String conversationId) {
|
||
return chatWithUsage(agentId, message, conversationId, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public ChatResult chatWithUsage(Long agentId, String message, String conversationId, ChatOrigin origin) {
|
||
return collectChatResult(chatStructuredStream(agentId, message, conversationId, "", null, origin));
|
||
}
|
||
|
||
public Flux<String> chatStream(Long agentId, String message, String conversationId) {
|
||
return chatStream(agentId, message, conversationId, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<String> chatStream(Long agentId, String message, String conversationId, ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, message);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
// Capture the origin into a request-scoped holder; cleared on Flux
|
||
// termination so the next reactive subscriber doesn't inherit stale state.
|
||
ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY;
|
||
return Flux.defer(() -> {
|
||
ChatOriginHolder.set(captured);
|
||
return withLifecycleFlux(agentId, message, conversationId,
|
||
(msg, convId) -> agent.chatStream(msg, convId),
|
||
chunk -> chunk);
|
||
}).doFinally(signal -> ChatOriginHolder.clear());
|
||
}
|
||
|
||
public Flux<StreamDelta> chatStructuredStream(Long agentId, String message, String conversationId) {
|
||
return chatStructuredStream(agentId, message, conversationId, "", null, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatStructuredStream(Long agentId, String message, String conversationId,
|
||
String requesterId) {
|
||
return chatStructuredStream(agentId, message, conversationId, requesterId, null, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatStructuredStream(Long agentId, String message, String conversationId,
|
||
String requesterId, ChatOrigin origin) {
|
||
return chatStructuredStream(agentId, message, conversationId, requesterId, null, origin);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatStructuredStream(Long agentId, String message, String conversationId,
|
||
String requesterId, String thinkingLevel) {
|
||
return chatStructuredStream(agentId, message, conversationId, requesterId, thinkingLevel,
|
||
ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatStructuredStream(Long agentId, String message, String conversationId,
|
||
String requesterId, String thinkingLevel,
|
||
ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, message);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
|
||
// 设置请求级思考深度（通过 ThreadLocal 传递到 StateGraph 执行）
|
||
if (thinkingLevel != null && !thinkingLevel.isBlank()) {
|
||
ThinkingLevelHolder.set(thinkingLevel);
|
||
} else {
|
||
// 尝试从 Agent 默认配置读取
|
||
AgentEntity entity = getAgent(agentId);
|
||
if (entity != null && entity.getDefaultThinkingLevel() != null) {
|
||
ThinkingLevelHolder.set(entity.getDefaultThinkingLevel());
|
||
} else {
|
||
ThinkingLevelHolder.clear();
|
||
}
|
||
}
|
||
|
||
ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY;
|
||
if (agent instanceof StructuredStreamCapable capable) {
|
||
return Flux.defer(() -> {
|
||
ChatOriginHolder.set(captured);
|
||
return withLifecycleFlux(agentId, message, conversationId,
|
||
(msg, convId) -> capable.chatStructuredStream(msg, convId,
|
||
requesterId != null ? requesterId : "")
|
||
.doFinally(signal -> ThinkingLevelHolder.clear()),
|
||
StreamDelta::content);
|
||
})
|
||
.doFinally(signal -> ChatOriginHolder.clear());
|
||
}
|
||
|
||
// 降级：不支持结构化流的 Agent，包装为纯内容流
|
||
ThinkingLevelHolder.clear();
|
||
return Flux.defer(() -> {
|
||
ChatOriginHolder.set(captured);
|
||
return withLifecycleFlux(agentId, message, conversationId,
|
||
(msg, convId) -> agent.chatStream(msg, convId)
|
||
.map(chunk -> new StreamDelta(chunk, null)),
|
||
StreamDelta::content);
|
||
})
|
||
.doFinally(signal -> ChatOriginHolder.clear());
|
||
}
|
||
|
||
public String execute(Long agentId, String goal, String conversationId) {
|
||
return execute(agentId, goal, conversationId, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public String execute(Long agentId, String goal, String conversationId, ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, goal);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY);
|
||
try {
|
||
return withLifecycleSync(agentId, goal, conversationId,
|
||
(msg, convId) -> agent.execute(msg, convId));
|
||
} finally {
|
||
ChatOriginHolder.clear();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 带工具重放的 chat 调用（审批通过后由 ChannelMessageRouter 或 ApprovalController 调用）
|
||
*
|
||
* @param agentId Agent ID
|
||
* @param userMessage 用户消息（如"继续执行已批准的工具"）
|
||
* @param conversationId 会话 ID
|
||
* @param toolCallPayload 要重放的工具调用 JSON
|
||
* @return Agent 回复
|
||
*/
|
||
public String chatWithReplay(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload) {
|
||
return chatWithReplay(agentId, userMessage, conversationId, toolCallPayload, ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public String chatWithReplay(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload, ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, userMessage);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
ChatOriginHolder.set(origin != null ? origin : ChatOrigin.EMPTY);
|
||
try {
|
||
return withLifecycleSync(agentId, userMessage, conversationId,
|
||
(msg, convId) -> agent.chatWithReplay(msg, convId, toolCallPayload));
|
||
} finally {
|
||
ChatOriginHolder.clear();
|
||
}
|
||
}
|
||
|
||
/**
|
||
* Replay-after-approval that also captures token usage and runtime model
|
||
* attribution. Mirrors {@link #chatWithUsage} for the
|
||
* approval-resumption path used by {@code ChannelMessageRouter}.
|
||
*/
|
||
public ChatResult chatWithReplayWithUsage(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload, ChatOrigin origin) {
|
||
return collectChatResult(chatWithReplayStream(agentId, userMessage, conversationId,
|
||
toolCallPayload, "", origin != null ? origin : ChatOrigin.EMPTY));
|
||
}
|
||
|
||
/**
|
||
* Subscribe to a structured stream and collapse it into a single
|
||
* {@link ChatResult}: append all content deltas, capture the trailing
|
||
* {@code _usage_final} event for token and model attribution.
|
||
*/
|
||
private ChatResult collectChatResult(Flux<StreamDelta> stream) {
|
||
StringBuilder content = new StringBuilder();
|
||
final int[] usage = {0, 0};
|
||
final String[] modelInfo = {null, null};
|
||
stream.doOnNext(delta -> {
|
||
if (delta.isEvent() && "_usage_final".equals(delta.eventType())) {
|
||
Map<String, Object> data = delta.eventData();
|
||
usage[0] = ((Number) data.getOrDefault("promptTokens", 0)).intValue();
|
||
usage[1] = ((Number) data.getOrDefault("completionTokens", 0)).intValue();
|
||
Object model = data.get("runtimeModelName");
|
||
Object provider = data.get("runtimeProviderId");
|
||
if (model != null) modelInfo[0] = model.toString();
|
||
if (provider != null) modelInfo[1] = provider.toString();
|
||
} else if (delta.content() != null) {
|
||
content.append(delta.content());
|
||
}
|
||
}).blockLast(Duration.ofMinutes(10));
|
||
return new ChatResult(content.toString(), usage[0], usage[1], modelInfo[0], modelInfo[1]);
|
||
}
|
||
|
||
/**
|
||
* 带工具重放的流式调用（Web 端审批通过后使用，通过 SSE 推送结果）
|
||
*/
|
||
public Flux<StreamDelta> chatWithReplayStream(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload) {
|
||
return chatWithReplayStream(agentId, userMessage, conversationId, toolCallPayload, "", ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatWithReplayStream(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload, String requesterId) {
|
||
return chatWithReplayStream(agentId, userMessage, conversationId, toolCallPayload, requesterId,
|
||
ChatOrigin.EMPTY);
|
||
}
|
||
|
||
public Flux<StreamDelta> chatWithReplayStream(Long agentId, String userMessage, String conversationId,
|
||
String toolCallPayload, String requesterId,
|
||
ChatOrigin origin) {
|
||
memoryRecallTracker.trackRecalls(agentId, userMessage);
|
||
BaseAgent agent = getOrBuildAgentForConversation(agentId, conversationId);
|
||
ChatOrigin captured = origin != null ? origin : ChatOrigin.EMPTY;
|
||
return Flux.defer(() -> {
|
||
ChatOriginHolder.set(captured);
|
||
return withLifecycleFlux(agentId, userMessage, conversationId,
|
||
(msg, convId) -> agent.chatWithReplayStream(msg, convId, toolCallPayload,
|
||
requesterId != null ? requesterId : ""),
|
||
StreamDelta::content);
|
||
})
|
||
.doFinally(signal -> ChatOriginHolder.clear());
|
||
}
|
||
|
||
public AgentState getAgentState(Long agentId) {
|
||
Map<String, BaseAgent> variants = agentInstances.get(agentId);
|
||
if (variants == null || variants.isEmpty()) {
|
||
return AgentState.IDLE;
|
||
}
|
||
// An Agent may have several cached graph variants (one per pinned
|
||
// model). Report the first non-IDLE state so a turn running on any
|
||
// variant stays visible.
|
||
for (BaseAgent agent : variants.values()) {
|
||
AgentState state = agent.getState();
|
||
if (state != AgentState.IDLE) {
|
||
return state;
|
||
}
|
||
}
|
||
return AgentState.IDLE;
|
||
}
|
||
|
||
// ==================== 缓存管理 ====================
|
||
|
||
public void refreshAgent(Long agentId) {
|
||
agentInstances.remove(agentId);
|
||
log.info("Agent instance cache cleared: {}", agentId);
|
||
}
|
||
|
||
public void refreshAllAgents() {
|
||
agentInstances.clear();
|
||
log.info("All agent instance caches cleared");
|
||
}
|
||
|
||
@EventListener
|
||
public void onModelConfigChanged(ModelConfigChangedEvent event) {
|
||
refreshAllAgents();
|
||
log.info("Agent caches refreshed after model config change: {}", event.reason());
|
||
}
|
||
|
||
@EventListener
|
||
public void onToolGuardConfigChanged(vip.mate.tool.guard.service.ToolGuardConfigService.ToolGuardConfigChangedEvent event) {
|
||
refreshAllAgents();
|
||
log.info("Agent caches refreshed after tool guard config change (denied tools may have changed)");
|
||
}
|
||
|
||
// ==================== Lifecycle helpers ====================
|
||
|
||
/**
|
||
* Wraps a synchronous agent call with lifecycle mediator hooks.
|
||
* When lifecycleMediatorEnabled is off, runs plainInvoke directly (Phase 0 behavior).
|
||
*
|
||
* P1-1 fix: prefetchAll result is now prepended to userMessage as <memory-context> block.
|
||
* P1-4 fix: N/A for sync (no cancel/error signal issue).
|
||
*/
|
||
private String withLifecycleSync(Long agentId, String message, String conversationId,
|
||
java.util.function.BiFunction<String, String, String> invoke) {
|
||
if (!memoryProperties.isLifecycleMediatorEnabled()) {
|
||
return invoke.apply(message, conversationId);
|
||
}
|
||
String ownerKey = memoryOwnerResolver.resolve(ChatOriginHolder.get());
|
||
TurnContext ctx = new TurnContext(agentId, conversationId, conversationId, 0, message, ownerKey);
|
||
String memoryContext = lifecycleMediator.beforeLlmCall(ctx);
|
||
// Inject memory context into the user message (RFC-037 §3.3)
|
||
String enrichedMessage = injectMemoryContext(message, memoryContext);
|
||
String result = invoke.apply(enrichedMessage, conversationId);
|
||
lifecycleMediator.afterLlmCall(ctx, result != null ? result : "");
|
||
return result;
|
||
}
|
||
|
||
/**
|
||
* Wraps a streaming agent call with lifecycle mediator hooks.
|
||
* When lifecycleMediatorEnabled is off, runs plainInvoke directly (Phase 0 behavior).
|
||
*
|
||
* P1-1 fix: prefetchAll result is now prepended to userMessage.
|
||
* P1-4 fix: afterLlmCall only fires on COMPLETE signal, not on cancel/error.
|
||
*/
|
||
private <T> Flux<T> withLifecycleFlux(Long agentId, String message, String conversationId,
|
||
java.util.function.BiFunction<String, String, Flux<T>> invoke,
|
||
Function<T, String> contentExtractor) {
|
||
if (!memoryProperties.isLifecycleMediatorEnabled()) {
|
||
return invoke.apply(message, conversationId);
|
||
}
|
||
String ownerKey = memoryOwnerResolver.resolve(ChatOriginHolder.get());
|
||
TurnContext ctx = new TurnContext(agentId, conversationId, conversationId, 0, message, ownerKey);
|
||
String memoryContext = lifecycleMediator.beforeLlmCall(ctx);
|
||
String enrichedMessage = injectMemoryContext(message, memoryContext);
|
||
StringBuilder reply = new StringBuilder();
|
||
return invoke.apply(enrichedMessage, conversationId)
|
||
.doOnNext(item -> {
|
||
String text = contentExtractor.apply(item);
|
||
if (text != null) {
|
||
reply.append(text);
|
||
}
|
||
})
|
||
.doOnComplete(() -> lifecycleMediator.afterLlmCall(ctx, reply.toString()))
|
||
.doOnError(e -> log.debug("[Memory] Stream error, skipping afterLlmCall: {}", e.getMessage()));
|
||
}
|
||
|
||
/**
|
||
* Prepend memory-context block to user message if non-empty.
|
||
* Does not pollute build-time system prompt snapshot.
|
||
*/
|
||
private String injectMemoryContext(String message, String memoryContext) {
|
||
if (memoryContext == null || memoryContext.isBlank()) return message;
|
||
return memoryContext + "\n\n" + message;
|
||
}
|
||
|
||
// ==================== 内部方法 ====================
|
||
|
||
/**
|
||
* Resolve (and cache) the Agent graph for a conversation, honouring the
|
||
* conversation's pinned model. Conversations with no pin — IM channels
|
||
* before issue #183 fix, cron, sub-tasks, or rows not yet created —
|
||
* resolve to the shared Agent / global-default graph.
|
||
*
|
||
* <p>Defensive normalisation: a half-populated pair (provider but no
|
||
* model, or vice versa) is treated as unpinned. Without this guard, a
|
||
* partially-cleared admin UI write could end up cached as a key like
|
||
* {@code "volcano::"} which {@link #getOrBuildAgent} would then try to
|
||
* build, only to fail at provider-resolution time on every turn.
|
||
*/
|
||
private BaseAgent getOrBuildAgentForConversation(Long agentId, String conversationId) {
|
||
String provider = null;
|
||
String modelName = null;
|
||
if (conversationId != null && !conversationId.isBlank()) {
|
||
ConversationEntity conv = conversationMapper.selectOne(
|
||
new LambdaQueryWrapper<ConversationEntity>()
|
||
.eq(ConversationEntity::getConversationId, conversationId));
|
||
if (conv != null) {
|
||
provider = blankToNull(conv.getModelProvider());
|
||
modelName = blankToNull(conv.getModelName());
|
||
// Half-populated pair → treat as unpinned. Pinning requires
|
||
// a complete (provider, model) tuple — see #183 follow-up
|
||
// hardening so a stale row written by an earlier broken
|
||
// admin UI release doesn't loop the cache on an invalid key.
|
||
if (provider == null || modelName == null) {
|
||
provider = null;
|
||
modelName = null;
|
||
}
|
||
}
|
||
}
|
||
return getOrBuildAgent(agentId, provider, modelName);
|
||
}
|
||
|
||
/** Map empty / whitespace strings to null so the pinned-check is one branch. */
|
||
private static String blankToNull(String s) {
|
||
return (s == null || s.isBlank()) ? null : s;
|
||
}
|
||
|
||
private BaseAgent getOrBuildAgent(Long agentId) {
|
||
return getOrBuildAgent(agentId, null, null);
|
||
}
|
||
|
||
private BaseAgent getOrBuildAgent(Long agentId, String modelProvider, String modelName) {
|
||
boolean pinned = modelProvider != null && !modelProvider.isBlank()
|
||
&& modelName != null && !modelName.isBlank();
|
||
String modelKey = pinned ? modelProvider + "::" + modelName : "";
|
||
return agentInstances
|
||
.computeIfAbsent(agentId, id -> new ConcurrentHashMap<>())
|
||
.computeIfAbsent(modelKey, key -> {
|
||
AgentEntity entity = getAgent(agentId);
|
||
if (!Boolean.TRUE.equals(entity.getEnabled())) {
|
||
throw new MateClawException("err.agent.disabled", "Agent 已禁用: " + entity.getName());
|
||
}
|
||
return agentGraphBuilder.build(entity, modelProvider, modelName);
|
||
});
|
||
}
|
||
|
||
// ==================== StreamDelta ====================
|
||
|
||
public record StreamDelta(String content, String thinking, String eventType, Map<String, Object> eventData,
|
||
boolean persistenceOnly, boolean segmentOnly) {
|
||
|
||
// 兼容构造器（广播+持久化）
|
||
public StreamDelta(String content, String thinking) {
|
||
this(content, thinking, null, null, false, false);
|
||
}
|
||
|
||
// 显式 5-参构造器：保留旧调用点对 (content, thinking, eventType, eventData, persistenceOnly) 的兼容
|
||
public StreamDelta(String content, String thinking, String eventType,
|
||
Map<String, Object> eventData, boolean persistenceOnly) {
|
||
this(content, thinking, eventType, eventData, persistenceOnly, false);
|
||
}
|
||
|
||
/** 仅用于持久化，不再广播（内容已由 NodeStreamingChatHelper 实时广播过） */
|
||
public static StreamDelta persistOnly(String content, String thinking) {
|
||
return new StreamDelta(content, thinking, null, null, true, false);
|
||
}
|
||
|
||
/**
|
||
* Per-iteration narrative routing for ReasoningNode / SummarizingNode output.
|
||
*
|
||
* <p>The accumulator should:
|
||
* <ul>
|
||
* <li>append the text to the in-flight {@code segments} entry so the UI's
|
||
* segmented view still renders the intermediate "I'll look it up…"
|
||
* narration between tool cards;</li>
|
||
* <li>NOT broadcast — already broadcast live by NodeStreamingChatHelper;</li>
|
||
* <li>NOT append to the top-level {@code content} StringBuilder, which is
|
||
* what gets persisted as {@code mate_message.content}. That field
|
||
* should hold the final-answer span only — otherwise multiple
|
||
* iterations stack into "我来…让我…然后…" walls that next-turn replay
|
||
* sees as unanswered chain-of-thought (issue #120 narration leg).</li>
|
||
* </ul>
|
||
*
|
||
* <p>Implies {@code persistenceOnly} (no broadcast) at the accumulator
|
||
* layer, but is a stricter promise: <em>nothing</em> reaches the top-level
|
||
* persisted content field via this flavor.
|
||
*/
|
||
public static StreamDelta segmentOnly(String content, String thinking) {
|
||
return new StreamDelta(content, thinking, null, null, true, true);
|
||
}
|
||
|
||
public static StreamDelta empty() {
|
||
return new StreamDelta(null, null, null, null, false, false);
|
||
}
|
||
|
||
public static StreamDelta event(String type, Map<String, Object> data) {
|
||
return new StreamDelta(null, null, type, data, false, false);
|
||
}
|
||
|
||
public boolean isEvent() {
|
||
return eventType != null;
|
||
}
|
||
|
||
public boolean hasPayload() {
|
||
return StringUtils.hasText(content) || StringUtils.hasText(thinking);
|
||
}
|
||
|
||
public int contentLength() {
|
||
return content != null ? content.length() : 0;
|
||
}
|
||
|
||
public int thinkingLength() {
|
||
return thinking != null ? thinking.length() : 0;
|
||
}
|
||
}
|
||
|
||
// ==================== ChatResult ====================
|
||
|
||
/**
|
||
* Sync chat result carrying the assistant reply alongside the usage
|
||
* attribution that the streaming path exposes via the {@code _usage_final}
|
||
* event. Use this when callers need to persist {@code promptTokens} /
|
||
* {@code completionTokens} / {@code runtimeModel} / {@code runtimeProvider}
|
||
* on the assistant message row but cannot subscribe to the structured
|
||
* stream directly (cron tasks, sync HTTP endpoints, voice WebSocket,
|
||
* post-approval replays).
|
||
*/
|
||
public record ChatResult(String content, int promptTokens, int completionTokens,
|
||
String runtimeModel, String runtimeProvider) {
|
||
|
||
public static ChatResult contentOnly(String content) {
|
||
return new ChatResult(content != null ? content : "", 0, 0, null, null);
|
||
}
|
||
}
|
||
}
|