mirror of
https://gitee.com/mateos/mateclaw.git
synced 2026-11-02 22:16:58 +08:00
fix(channels): expire stale query context in persistent conversations
This commit is contained in:
parent
845f7b46d4
commit
3fbe5a18a1
@ -182,6 +182,14 @@ public class AgentGraphBuilder {
|
|||||||
*/
|
*/
|
||||||
private vip.mate.audit.service.AuditEventService auditEventService;
|
private vip.mate.audit.service.AuditEventService auditEventService;
|
||||||
|
|
||||||
|
private vip.mate.config.ChannelHistoryProperties channelHistoryProperties =
|
||||||
|
new vip.mate.config.ChannelHistoryProperties();
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public void setChannelHistoryProperties(vip.mate.config.ChannelHistoryProperties properties) {
|
||||||
|
this.channelHistoryProperties = properties;
|
||||||
|
}
|
||||||
|
|
||||||
@Autowired(required = false)
|
@Autowired(required = false)
|
||||||
public void setAuditEventService(vip.mate.audit.service.AuditEventService s) {
|
public void setAuditEventService(vip.mate.audit.service.AuditEventService s) {
|
||||||
this.auditEventService = s;
|
this.auditEventService = s;
|
||||||
@ -506,6 +514,7 @@ public class AgentGraphBuilder {
|
|||||||
agent.agentId = String.valueOf(entity.getId());
|
agent.agentId = String.valueOf(entity.getId());
|
||||||
agent.agentName = entity.getName();
|
agent.agentName = entity.getName();
|
||||||
agent.systemPrompt = enhancedPrompt;
|
agent.systemPrompt = enhancedPrompt;
|
||||||
|
agent.channelHistoryPolicy = new vip.mate.agent.context.ChannelHistoryPolicy(channelHistoryProperties);
|
||||||
agent.maxIterations = maxIter;
|
agent.maxIterations = maxIter;
|
||||||
agent.modelName = runtimeModel.getModelName();
|
agent.modelName = runtimeModel.getModelName();
|
||||||
agent.modelCapabilities = modelCapabilityService.resolve(
|
agent.modelCapabilities = modelCapabilityService.resolve(
|
||||||
|
|||||||
@ -59,6 +59,9 @@ public abstract class BaseAgent {
|
|||||||
/** 系统提示词 */
|
/** 系统提示词 */
|
||||||
protected String systemPrompt;
|
protected String systemPrompt;
|
||||||
|
|
||||||
|
protected vip.mate.agent.context.ChannelHistoryPolicy channelHistoryPolicy =
|
||||||
|
new vip.mate.agent.context.ChannelHistoryPolicy(new vip.mate.config.ChannelHistoryProperties());
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Max ReAct iterations (one reasoning + action + observation step counts as one).
|
* Max ReAct iterations (one reasoning + action + observation step counts as one).
|
||||||
* Default 150, hard ceiling 150 (enforced in AgentGraphBuilder so per-agent DB
|
* Default 150, hard ceiling 150 (enforced in AgentGraphBuilder so per-agent DB
|
||||||
@ -258,6 +261,8 @@ public abstract class BaseAgent {
|
|||||||
// The gate is an explicit ChatOrigin signal, so a normal Web or
|
// The gate is an explicit ChatOrigin signal, so a normal Web or
|
||||||
// channel turn can never take this path.
|
// channel turn can never take this path.
|
||||||
ChatOrigin chatOrigin = ChatOriginHolder.get();
|
ChatOrigin chatOrigin = ChatOriginHolder.get();
|
||||||
|
boolean channelHistory = channelHistoryPolicy.applies(chatOrigin);
|
||||||
|
java.time.LocalDateTime historyNow = java.time.LocalDateTime.now();
|
||||||
if (chatOrigin != null && chatOrigin.cronOrigin()) {
|
if (chatOrigin != null && chatOrigin.cronOrigin()) {
|
||||||
log.info("[{}] Scheduled-job run: LLM context isolated (no conversation history replayed)",
|
log.info("[{}] Scheduled-job run: LLM context isolated (no conversation history replayed)",
|
||||||
agentName);
|
agentName);
|
||||||
@ -267,7 +272,7 @@ public abstract class BaseAgent {
|
|||||||
// ===== 两阶段加载:短对话全量,长对话分页(递进式) =====
|
// ===== 两阶段加载:短对话全量,长对话分页(递进式) =====
|
||||||
long totalCount = conversationService.countMessages(conversationId);
|
long totalCount = conversationService.countMessages(conversationId);
|
||||||
if (totalCount <= 0) {
|
if (totalCount <= 0) {
|
||||||
return List.of();
|
return channelHistory ? List.of(new SystemMessage(channelHistoryPolicy.guidance(historyNow))) : List.of();
|
||||||
}
|
}
|
||||||
|
|
||||||
int windowSize = getEffectiveWindowSize();
|
int windowSize = getEffectiveWindowSize();
|
||||||
@ -292,7 +297,7 @@ public abstract class BaseAgent {
|
|||||||
boolean boundaryFoundInWindow = false;
|
boolean boundaryFoundInWindow = false;
|
||||||
for (int i = history.size() - 1; i >= 0; i--) {
|
for (int i = history.size() - 1; i >= 0; i--) {
|
||||||
MessageEntity msg = history.get(i);
|
MessageEntity msg = history.get(i);
|
||||||
if ("system".equals(msg.getRole()) && isCompressionSummary(msg)) {
|
if (!channelHistory && "system".equals(msg.getRole()) && isCompressionSummary(msg)) {
|
||||||
history = new ArrayList<>(history.subList(i, history.size()));
|
history = new ArrayList<>(history.subList(i, history.size()));
|
||||||
boundaryFoundInWindow = true;
|
boundaryFoundInWindow = true;
|
||||||
log.info("[{}] Found latest compression boundary at index {}; loading {} messages forward",
|
log.info("[{}] Found latest compression boundary at index {}; loading {} messages forward",
|
||||||
@ -308,7 +313,7 @@ public abstract class BaseAgent {
|
|||||||
// original list and never made it into `history`. Without prepending
|
// original list and never made it into `history`. Without prepending
|
||||||
// it, the model would forget the original goal even though we already
|
// it, the model would forget the original goal even though we already
|
||||||
// paid the LLM cost to produce a structured summary.
|
// paid the LLM cost to produce a structured summary.
|
||||||
if (!boundaryFoundInWindow && totalCount > windowSize) {
|
if (!channelHistory && !boundaryFoundInWindow && totalCount > windowSize) {
|
||||||
try {
|
try {
|
||||||
MessageEntity latestBoundary = conversationService.findLatestCompressionBoundary(conversationId);
|
MessageEntity latestBoundary = conversationService.findLatestCompressionBoundary(conversationId);
|
||||||
if (latestBoundary != null) {
|
if (latestBoundary != null) {
|
||||||
@ -332,11 +337,17 @@ public abstract class BaseAgent {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (channelHistory) {
|
||||||
|
history = channelHistoryPolicy.select(history.subList(0, limit), historyNow);
|
||||||
|
limit = history.size();
|
||||||
|
}
|
||||||
|
|
||||||
if (limit <= 0) {
|
if (limit <= 0) {
|
||||||
return List.of();
|
return channelHistory ? List.of(new SystemMessage(channelHistoryPolicy.guidance(historyNow))) : List.of();
|
||||||
}
|
}
|
||||||
|
|
||||||
List<Message> messages = new ArrayList<>(limit);
|
List<Message> messages = new ArrayList<>(limit);
|
||||||
|
if (channelHistory) messages.add(new SystemMessage(channelHistoryPolicy.guidance(historyNow)));
|
||||||
for (int i = 0; i < limit; i += 1) {
|
for (int i = 0; i < limit; i += 1) {
|
||||||
messages.addAll(expandToSpringMessages(history.get(i)));
|
messages.addAll(expandToSpringMessages(history.get(i)));
|
||||||
}
|
}
|
||||||
@ -894,6 +905,9 @@ public abstract class BaseAgent {
|
|||||||
renderedContent = directToolHistoryPlaceholder(directNames);
|
renderedContent = directToolHistoryPlaceholder(directNames);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (channelHistoryPolicy.applies(ChatOriginHolder.get()) && message.getCreateTime() != null) {
|
||||||
|
renderedContent = "[Historical message at " + message.getCreateTime() + "]\n" + renderedContent;
|
||||||
|
}
|
||||||
return switch (message.getRole()) {
|
return switch (message.getRole()) {
|
||||||
case "assistant" -> new AssistantMessage(renderedContent);
|
case "assistant" -> new AssistantMessage(renderedContent);
|
||||||
case "system" -> isCompressionSummary(message)
|
case "system" -> isCompressionSummary(message)
|
||||||
@ -1330,6 +1344,8 @@ public abstract class BaseAgent {
|
|||||||
for (int j = currentIdx - 1; j >= from; j--) {
|
for (int j = currentIdx - 1; j >= from; j--) {
|
||||||
MessageEntity m = history.get(j);
|
MessageEntity m = history.get(j);
|
||||||
if (m == null || !"user".equals(m.getRole())) continue;
|
if (m == null || !"user".equals(m.getRole())) continue;
|
||||||
|
if (channelHistoryPolicy.applies(ChatOriginHolder.get())
|
||||||
|
&& !channelHistoryPolicy.isHot(m, java.time.LocalDateTime.now())) continue;
|
||||||
List<MessageContentPart> parts = conversationService.parseMessageParts(m);
|
List<MessageContentPart> parts = conversationService.parseMessageParts(m);
|
||||||
for (int k = parts.size() - 1; k >= 0; k--) {
|
for (int k = parts.size() - 1; k >= 0; k--) {
|
||||||
MessageContentPart part = parts.get(k);
|
MessageContentPart part = parts.get(k);
|
||||||
|
|||||||
@ -0,0 +1,94 @@
|
|||||||
|
package vip.mate.agent.context;
|
||||||
|
|
||||||
|
import vip.mate.config.ChannelHistoryProperties;
|
||||||
|
import vip.mate.workspace.conversation.model.MessageEntity;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/** Non-destructive projection of persisted channel history into model context. */
|
||||||
|
public class ChannelHistoryPolicy {
|
||||||
|
private final ChannelHistoryProperties properties;
|
||||||
|
|
||||||
|
public ChannelHistoryPolicy(ChannelHistoryProperties properties) {
|
||||||
|
this.properties = properties;
|
||||||
|
}
|
||||||
|
|
||||||
|
public boolean applies(ChatOrigin origin) {
|
||||||
|
return properties.isEnabled() && origin != null && !origin.cronOrigin()
|
||||||
|
&& origin.channelType() != null && !origin.channelType().isBlank()
|
||||||
|
&& !"web".equals(origin.channelType());
|
||||||
|
}
|
||||||
|
|
||||||
|
public String guidance(LocalDateTime now) {
|
||||||
|
return "[Channel history freshness; current server local time: " + now + "] "
|
||||||
|
+ "Conversation history, summaries and recalled memories are historical context, not current evidence. "
|
||||||
|
+ "For current/latest/today's business data, query the authoritative source in this turn, "
|
||||||
|
+ "even if a recent answer exists. Report the query time and data period. "
|
||||||
|
+ "If querying fails or is unavailable, say current data could not be verified; never substitute old values. "
|
||||||
|
+ "Use historical snapshots only when explicitly requested and label their time. "
|
||||||
|
+ "Older answers and tool results have been withheld; retained old questions are reference context only.";
|
||||||
|
}
|
||||||
|
|
||||||
|
public boolean isHot(MessageEntity row, LocalDateTime now) {
|
||||||
|
return row.getCreateTime() != null && !row.getCreateTime().isAfter(now)
|
||||||
|
&& !row.getCreateTime().isBefore(now.minusMinutes(Math.max(0, properties.getHotMinutes())));
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Input is chronological, excluding the current user row. Age whole turns
|
||||||
|
* together so a cutoff never separates an assistant tool call from its result.
|
||||||
|
* Summary timestamps describe compression time, not the age of their facts:
|
||||||
|
* never let a newly generated summary make old query results fresh again.
|
||||||
|
*/
|
||||||
|
public List<MessageEntity> select(List<MessageEntity> rows, LocalDateTime now) {
|
||||||
|
List<MessageEntity> result = new ArrayList<>();
|
||||||
|
List<MessageEntity> turn = new ArrayList<>();
|
||||||
|
for (MessageEntity row : rows) {
|
||||||
|
if ("system".equals(row.getRole())) continue;
|
||||||
|
if ("user".equals(row.getRole()) && !turn.isEmpty()) {
|
||||||
|
appendTurn(result, turn, now);
|
||||||
|
turn.clear();
|
||||||
|
}
|
||||||
|
turn.add(row);
|
||||||
|
}
|
||||||
|
appendTurn(result, turn, now);
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
|
||||||
|
private void appendTurn(List<MessageEntity> result, List<MessageEntity> turn, LocalDateTime now) {
|
||||||
|
if (turn.isEmpty()) return;
|
||||||
|
LocalDateTime oldest = now;
|
||||||
|
for (MessageEntity row : turn) {
|
||||||
|
// Legacy rows without provenance cannot be classified as fresh.
|
||||||
|
if (row.getCreateTime() == null || row.getCreateTime().isAfter(now)) return;
|
||||||
|
if (row.getCreateTime().isBefore(oldest)) oldest = row.getCreateTime();
|
||||||
|
}
|
||||||
|
long hot = Math.max(0, properties.getHotMinutes());
|
||||||
|
long warm = Math.max(hot, properties.getWarmMinutes());
|
||||||
|
if (oldest.isBefore(now.minusMinutes(warm))) return;
|
||||||
|
if (!oldest.isBefore(now.minusMinutes(hot))) {
|
||||||
|
result.addAll(turn);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
// Do not retain assistant paraphrases of stale results, tool arguments,
|
||||||
|
// structured content parts or metadata containing the same payload.
|
||||||
|
for (MessageEntity row : turn) {
|
||||||
|
if (!"user".equals(row.getRole()) || row.getContent() == null || row.getContent().isBlank()) continue;
|
||||||
|
result.add(textRow(row, "user", "[Historical request at " + row.getCreateTime()
|
||||||
|
+ "; not current data]\n" + row.getContent()));
|
||||||
|
result.add(textRow(row, "assistant", "[Previous answer and tool results expired. "
|
||||||
|
+ "Query the source again if needed; do not reconstruct old values.]"));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MessageEntity textRow(MessageEntity original, String role, String text) {
|
||||||
|
MessageEntity row = new MessageEntity();
|
||||||
|
row.setRole(role);
|
||||||
|
row.setContent(text);
|
||||||
|
row.setStatus("completed");
|
||||||
|
row.setCreateTime(original.getCreateTime());
|
||||||
|
return row;
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -5,6 +5,9 @@ import com.fasterxml.jackson.core.JsonProcessingException;
|
|||||||
import com.fasterxml.jackson.databind.ObjectMapper;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import vip.mate.agent.context.ChannelHistoryPolicy;
|
||||||
|
import vip.mate.config.ChannelHistoryProperties;
|
||||||
import vip.mate.agent.context.ChatOrigin;
|
import vip.mate.agent.context.ChatOrigin;
|
||||||
import vip.mate.agent.context.TokenEstimator;
|
import vip.mate.agent.context.TokenEstimator;
|
||||||
import vip.mate.workspace.conversation.model.MessageEntity;
|
import vip.mate.workspace.conversation.model.MessageEntity;
|
||||||
@ -27,15 +30,24 @@ public class DshConversationHistory {
|
|||||||
private final MessageMapper mapper;
|
private final MessageMapper mapper;
|
||||||
private final ObjectMapper objectMapper;
|
private final ObjectMapper objectMapper;
|
||||||
|
|
||||||
|
private ChannelHistoryPolicy channelHistoryPolicy = new ChannelHistoryPolicy(new ChannelHistoryProperties());
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public void setChannelHistoryProperties(ChannelHistoryProperties properties) {
|
||||||
|
channelHistoryPolicy = new ChannelHistoryPolicy(properties);
|
||||||
|
}
|
||||||
|
|
||||||
public String enrich(String conversationId, String originalInput, String currentInput, ChatOrigin origin) {
|
public String enrich(String conversationId, String originalInput, String currentInput, ChatOrigin origin) {
|
||||||
if (conversationId == null || conversationId.isBlank() || (origin != null && origin.cronOrigin())) {
|
if (conversationId == null || conversationId.isBlank() || (origin != null && origin.cronOrigin())) {
|
||||||
return currentInput;
|
return currentInput;
|
||||||
}
|
}
|
||||||
Long beforeId = origin == null ? null : origin.originMessageId();
|
Long beforeId = origin == null ? null : origin.originMessageId();
|
||||||
|
boolean channelHistory = channelHistoryPolicy.applies(origin);
|
||||||
|
java.time.LocalDateTime now = java.time.LocalDateTime.now();
|
||||||
// A leaf mapper avoids a circular dependency through ConversationService.
|
// A leaf mapper avoids a circular dependency through ConversationService.
|
||||||
// Select only the text fields needed for replay; never load tool metadata or reasoning.
|
// Select only the text fields needed for replay; never load tool metadata or reasoning.
|
||||||
List<MessageEntity> rows = mapper.selectList(new LambdaQueryWrapper<MessageEntity>()
|
List<MessageEntity> rows = mapper.selectList(new LambdaQueryWrapper<MessageEntity>()
|
||||||
.select(MessageEntity::getId, MessageEntity::getRole, MessageEntity::getContent)
|
.select(MessageEntity::getId, MessageEntity::getRole, MessageEntity::getContent, MessageEntity::getCreateTime)
|
||||||
.eq(MessageEntity::getConversationId, conversationId)
|
.eq(MessageEntity::getConversationId, conversationId)
|
||||||
.eq(MessageEntity::getDeleted, 0)
|
.eq(MessageEntity::getDeleted, 0)
|
||||||
.eq(MessageEntity::getStatus, "completed")
|
.eq(MessageEntity::getStatus, "completed")
|
||||||
@ -44,15 +56,22 @@ public class DshConversationHistory {
|
|||||||
.orderByDesc(MessageEntity::getId)
|
.orderByDesc(MessageEntity::getId)
|
||||||
.last("LIMIT " + MAX_MESSAGES));
|
.last("LIMIT " + MAX_MESSAGES));
|
||||||
|
|
||||||
|
// De-duplicate before projecting: warm rows are deliberately rewritten.
|
||||||
|
if (beforeId == null && !rows.isEmpty() && "user".equals(rows.getFirst().getRole())
|
||||||
|
&& java.util.Objects.equals(originalInput, rows.getFirst().getContent())) {
|
||||||
|
rows = new ArrayList<>(rows.subList(1, rows.size()));
|
||||||
|
}
|
||||||
|
if (channelHistory) {
|
||||||
|
rows = channelHistoryPolicy.select(rows.reversed(), now).reversed();
|
||||||
|
currentInput = channelHistoryPolicy.guidance(now) + "\n\n" + currentInput;
|
||||||
|
}
|
||||||
|
|
||||||
List<HistoricalMessage> selected = new ArrayList<>();
|
List<HistoricalMessage> selected = new ArrayList<>();
|
||||||
for (int index = 0; index < rows.size(); index++) {
|
for (int index = 0; index < rows.size(); index++) {
|
||||||
MessageEntity row = rows.get(index);
|
MessageEntity row = rows.get(index);
|
||||||
// Legacy callers may not supply an origin ID. Only drop the latest
|
|
||||||
// matching user row, preserving older intentionally repeated questions.
|
|
||||||
if (beforeId == null && index == 0 && "user".equals(row.getRole())
|
|
||||||
&& java.util.Objects.equals(originalInput, row.getContent())) continue;
|
|
||||||
String content = row.getContent();
|
String content = row.getContent();
|
||||||
if (content == null || content.isBlank()) continue;
|
if (content == null || content.isBlank()) continue;
|
||||||
|
if (channelHistory) content = "[Historical message at " + row.getCreateTime() + "]\n" + content;
|
||||||
selected.addFirst(new HistoricalMessage(row.getRole(), content));
|
selected.addFirst(new HistoricalMessage(row.getRole(), content));
|
||||||
if (TokenEstimator.estimateTokens(frame(selected)) <= MAX_HISTORY_TOKENS) continue;
|
if (TokenEstimator.estimateTokens(frame(selected)) <= MAX_HISTORY_TOKENS) continue;
|
||||||
|
|
||||||
|
|||||||
@ -0,0 +1,15 @@
|
|||||||
|
package vip.mate.config;
|
||||||
|
|
||||||
|
import lombok.Data;
|
||||||
|
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
/** Wall-clock history limits, independent of storage retention and token budgets. */
|
||||||
|
@Data
|
||||||
|
@Component
|
||||||
|
@ConfigurationProperties(prefix = "mate.agent.channel-history")
|
||||||
|
public class ChannelHistoryProperties {
|
||||||
|
private boolean enabled = true;
|
||||||
|
private long hotMinutes = 30;
|
||||||
|
private long warmMinutes = 1440;
|
||||||
|
}
|
||||||
@ -0,0 +1,69 @@
|
|||||||
|
# 渠道问数历史新鲜度治理
|
||||||
|
|
||||||
|
## 问题与根因
|
||||||
|
|
||||||
|
飞书、微信等渠道通常按渠道配置和聊天/发送者生成固定 conversationId。固定 ID 用于消息归属、推送和定时任务路由是合理的,但不能因此把该会话的所有历史当成当前数据。
|
||||||
|
|
||||||
|
此次源码排查发现:
|
||||||
|
|
||||||
|
1. `ChannelMessageRouter.buildConversationId` 没有时间维度,同一聊天长期复用会话。
|
||||||
|
2. `BaseAgent.buildConversationHistory` 原先只按消息数量分页、按压缩边界回放,未检查 `createTime`。分页还会补回窗口外的最近压缩摘要;旧查询数值可以通过旧答案和摘要重复进入模型。
|
||||||
|
3. `DshConversationHistory` 原先只限制最近 40 条、4096 token,没有时间过滤,且回放文本不带消息时间。
|
||||||
|
4. `ChannelSessionStore` 的 30 天 TTL 只在缓存超量时淘汰内存推送地址映射,不删除数据库记录,也不控制模型历史。
|
||||||
|
5. `ConversationWindowManager.compactAgedToolResponses` 的 age 指工具结果的相对顺序,不是实际经过的分钟数;压缩和 token 裁剪不能替代业务数据有效期。
|
||||||
|
6. 图片追问会自动携带最近若干条消息里的图片,原先也没有时间限制,可能带回旧报表。
|
||||||
|
|
||||||
|
这是根据源码和回归用例确认的上下文污染路径;没有读取生产渠道日志或连接生产数据源。
|
||||||
|
|
||||||
|
## QwenPaw 参考结论
|
||||||
|
|
||||||
|
参考本机 `/Users/mate/Codes/Open-Source/ai/QwenPaw`:
|
||||||
|
|
||||||
|
- `src/qwenpaw/app/channels/feishu/channel.py::resolve_session_id` 和 `wechat/channel.py::resolve_session_id` 同样使用稳定的发送者/群标识。
|
||||||
|
- `src/qwenpaw/app/chats/session.py::SafeJSONSession.load_session_state` 从 JSON 恢复状态时未在该入口判定业务数据是否过期。
|
||||||
|
- `ToolResultPruningConfig` 区分近期和较早的工具结果大小;`ScrollContextConfig.history_retention_days` 默认 30 天,管理持久历史保留。这些是上下文容量与存储管理机制,不能视为“查询数据在此时间内仍然有效”的承诺。
|
||||||
|
|
||||||
|
借鉴其持久历史与活动上下文分离的思路,在 MateClaw 修复模型入口。本次未修改 QwenPaw 仓库。
|
||||||
|
|
||||||
|
## 当前修复
|
||||||
|
|
||||||
|
StateGraph 与 DSH 共用 `ChannelHistoryPolicy`,仅改变本轮模型输入,不修改持久记录、会话 ID 或推送映射。默认覆盖有 `channelType` 的非 Web、非定时任务调用,包括飞书、微信、企业微信、钉钉及 webchat。
|
||||||
|
|
||||||
|
| 层级 | 默认时间 | 模型输入 |
|
||||||
|
| --- | --- | --- |
|
||||||
|
| 热 | 最近 30 分钟 | 保留完整轮次;对话文本标记原始消息时间,仍是历史证据 |
|
||||||
|
| 温 | 30 分钟至 24 小时 | 保留用户原问题供理解指代;旧助手答案、工具调用/结果和结构化载荷替换为过期提示 |
|
||||||
|
| 冷 | 超过 24 小时 | 不自动回放;原记录继续保留 |
|
||||||
|
|
||||||
|
时间边界按整个用户轮次中最早的消息时间判断,避免在工具调用与结果之间截断。缺失时间、未来时间的轮次不自动回放。所有渠道压缩摘要不再自动注入,因为摘要生成时间不能证明其引用的数据新鲜。此取舍会减少长会话跨天的自动续接能力;用户需重新提供必要条件,或明确请求查阅历史。
|
||||||
|
|
||||||
|
两条引擎都附加新鲜度规则:当前/最新/今日问数必须在本轮查询权威数据源,说明查询时间和数据期间;查询失败或不可用时说明无法核验,不用旧值代替。自动携带历史图片只允许热层图片。用户本轮主动提供的图片不受这个历史限制影响。
|
||||||
|
|
||||||
|
配置示例(分钟,服务端本地时间口径,与现有 `createTime` 一致):
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
mate:
|
||||||
|
agent:
|
||||||
|
channel-history:
|
||||||
|
enabled: true
|
||||||
|
hot-minutes: 30
|
||||||
|
warm-minutes: 1440
|
||||||
|
```
|
||||||
|
|
||||||
|
默认无需配置即可生效;重启运行更新后的服务端。`enabled: false` 可回到原历史回放行为。实现将负数按 0 处理,温层上限至少为热层上限;建议配置满足 `0 <= hot-minutes <= warm-minutes`。
|
||||||
|
|
||||||
|
## 验证与落地建议
|
||||||
|
|
||||||
|
回归覆盖冷热边界、旧结果及其答案移除、结构化载荷清除、摘要防回填、工具调用轮次完整性、原记录不变、DSH SQL 加载与当前消息去重、Web/定时任务兼容,以及历史图片的时间限制。
|
||||||
|
|
||||||
|
验证命令:
|
||||||
|
|
||||||
|
```sh
|
||||||
|
mvn -pl mateclaw-server -am \
|
||||||
|
-Dtest='ChannelHistoryPolicyTest,BaseAgent*Test,DshConversationHistoryTest' \
|
||||||
|
-Dsurefire.failIfNoSpecifiedTests=false test
|
||||||
|
```
|
||||||
|
|
||||||
|
上线验收建议:同一飞书/微信聊天先查询一次库存,改变测试数据后立即再问“现在库存多少”,再分别构造两小时前和两天前的历史;核查本轮工具日志确实重新查询、回答中的时间和数值正确。另测数据源不可用时是否明确说明无法核验。
|
||||||
|
|
||||||
|
30 分钟是对话保留窗口,不是业务数据缓存 TTL。本次硬性移除了过期历史结果,但“最新问数必查”的规则仍由模型执行,不能宣称是工具调用强制校验。对于库存、余额、实时经营数据,后续应按数据源 SLA 增加结构化 `queried_at / data_as_of / valid_until`,在回答出口验证本轮查询凭证;长期记忆只沉淀指标口径、表结构、用户偏好,瞬时数值应带时间和有效期。此次未改造长期记忆的存储与检索,也未实施生产发布。
|
||||||
@ -26,6 +26,32 @@ import static org.mockito.Mockito.when;
|
|||||||
*/
|
*/
|
||||||
class BaseAgentCarryRecentImageTest {
|
class BaseAgentCarryRecentImageTest {
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void channelDoesNotCarryExpiredReportButStillCarriesHotImage() throws Exception {
|
||||||
|
Path img = Files.createTempFile("carry-channel-report", ".jpg");
|
||||||
|
Files.write(img, new byte[]{(byte) 0xFF, (byte) 0xD8, (byte) 0xFF, 0x00});
|
||||||
|
try {
|
||||||
|
vip.mate.agent.context.ChatOriginHolder.set(vip.mate.agent.context.ChatOrigin.EMPTY
|
||||||
|
.withSender("u", "feishu", null));
|
||||||
|
TestAgent agent = visionAgent();
|
||||||
|
MessageEntity report = userMsg("库存报表");
|
||||||
|
report.setCreateTime(java.time.LocalDateTime.now().minusHours(2));
|
||||||
|
MessageEntity current = userMsg("现在库存多少");
|
||||||
|
when(agent.conversationService.listMessages("c1")).thenReturn(List.of(report, current));
|
||||||
|
when(agent.conversationService.renderMessageContent(current)).thenReturn("现在库存多少");
|
||||||
|
when(agent.conversationService.parseMessageParts(report))
|
||||||
|
.thenReturn(List.of(imagePart(img.toAbsolutePath().toString())));
|
||||||
|
when(agent.conversationService.parseMessageParts(current)).thenReturn(List.of());
|
||||||
|
UserMessage expired = agent.callBuildCurrent("c1", "现在库存多少");
|
||||||
|
assertTrue(expired.getMedia() == null || expired.getMedia().isEmpty());
|
||||||
|
report.setCreateTime(java.time.LocalDateTime.now().minusMinutes(1));
|
||||||
|
assertTrue(agent.callBuildCurrent("c1", "现在库存多少").getMedia().size() == 1);
|
||||||
|
} finally {
|
||||||
|
vip.mate.agent.context.ChatOriginHolder.clear();
|
||||||
|
Files.deleteIfExists(img);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
@DisplayName("Vision model + follow-up with no image → most recent image is carried into the turn")
|
@DisplayName("Vision model + follow-up with no image → most recent image is carried into the turn")
|
||||||
void followUp_carriesRecentImage() throws Exception {
|
void followUp_carriesRecentImage() throws Exception {
|
||||||
|
|||||||
@ -0,0 +1,71 @@
|
|||||||
|
package vip.mate.agent;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.AfterEach;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import org.springframework.ai.chat.messages.Message;
|
||||||
|
import vip.mate.agent.context.ChatOrigin;
|
||||||
|
import vip.mate.agent.context.ChatOriginHolder;
|
||||||
|
import vip.mate.workspace.conversation.ConversationService;
|
||||||
|
import vip.mate.workspace.conversation.model.MessageEntity;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
import static org.mockito.ArgumentMatchers.any;
|
||||||
|
import static org.mockito.Mockito.*;
|
||||||
|
|
||||||
|
class BaseAgentChannelHistoryTest {
|
||||||
|
@AfterEach void clear() { ChatOriginHolder.clear(); }
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void feishuDropsStaleResultsAndDoesNotReinjectOutOfWindowSummary() {
|
||||||
|
var conv = mock(ConversationService.class);
|
||||||
|
when(conv.countMessages("c")).thenReturn(1000L);
|
||||||
|
var question = row("user", "查询华北库存", 120);
|
||||||
|
var stale = row("assistant", "stale inventory 98765", 119);
|
||||||
|
var recent = row("user", "按照仓库分组", 5);
|
||||||
|
var answer = row("assistant", "recent answer", 4);
|
||||||
|
when(conv.listRecentMessages(eq("c"), anyInt())).thenReturn(List.of(question, stale, recent, answer));
|
||||||
|
when(conv.renderMessageContent(any())).thenAnswer(i -> ((MessageEntity) i.getArgument(0)).getContent());
|
||||||
|
ChatOriginHolder.set(ChatOrigin.EMPTY.withSender("u", "feishu", null));
|
||||||
|
var agent = new BaseAgentCronIsolationTest.TestAgent(conv);
|
||||||
|
String prompt = agent.history("c", "最新情况").stream().map(Message::getText).reduce("", (a, b) -> a + b);
|
||||||
|
assertFalse(prompt.contains("98765"));
|
||||||
|
assertTrue(prompt.contains("查询华北库存"));
|
||||||
|
assertTrue(prompt.contains("recent answer"));
|
||||||
|
assertTrue(prompt.contains("query the authoritative source in this turn"));
|
||||||
|
verify(conv, never()).findLatestCompressionBoundary(any());
|
||||||
|
assertEquals("stale inventory 98765", stale.getContent());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void channelDropsFreshlyWrittenCompressionSummaryAndDeduplicatesCurrentInput() {
|
||||||
|
var conv = mock(ConversationService.class);
|
||||||
|
var summary = row("system", "stale summarized 98765", 1);
|
||||||
|
summary.setMetadata("compression_summary");
|
||||||
|
when(conv.countMessages("c")).thenReturn(2L);
|
||||||
|
when(conv.listMessages("c")).thenReturn(List.of(summary, row("user", "current", 0)));
|
||||||
|
ChatOriginHolder.set(ChatOrigin.EMPTY.withSender("u", "weixin", null));
|
||||||
|
var result = new BaseAgentCronIsolationTest.TestAgent(conv).history("c", "current");
|
||||||
|
assertEquals(1, result.size());
|
||||||
|
assertFalse(result.getFirst().getText().contains("98765"));
|
||||||
|
verify(conv, never()).renderMessageContent(any());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void firstTurnStillGetsFreshnessGuidance() {
|
||||||
|
var conv = mock(ConversationService.class);
|
||||||
|
ChatOriginHolder.set(ChatOrigin.EMPTY.withSender("u", "weixin", null));
|
||||||
|
assertEquals(1, new BaseAgentCronIsolationTest.TestAgent(conv).history("c", "最新库存").size());
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MessageEntity row(String role, String text, long minutesAgo) {
|
||||||
|
var row = new MessageEntity();
|
||||||
|
row.setRole(role);
|
||||||
|
row.setContent(text);
|
||||||
|
row.setCreateTime(LocalDateTime.now().minusMinutes(minutesAgo));
|
||||||
|
row.setStatus("completed");
|
||||||
|
return row;
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -0,0 +1,87 @@
|
|||||||
|
package vip.mate.agent.context;
|
||||||
|
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import vip.mate.config.ChannelHistoryProperties;
|
||||||
|
import vip.mate.workspace.conversation.model.MessageEntity;
|
||||||
|
|
||||||
|
import java.time.LocalDateTime;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.junit.jupiter.api.Assertions.*;
|
||||||
|
|
||||||
|
class ChannelHistoryPolicyTest {
|
||||||
|
private final LocalDateTime now = LocalDateTime.of(2026, 9, 21, 12, 0);
|
||||||
|
private final ChannelHistoryProperties properties = new ChannelHistoryProperties();
|
||||||
|
private final ChannelHistoryPolicy policy = new ChannelHistoryPolicy(properties);
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void hotWarmColdProjectionDoesNotMutateStoredData() {
|
||||||
|
var cold = row("user", "cold question", now.minusDays(2));
|
||||||
|
var warm = row("user", "华北库存是多少", now.minusHours(2));
|
||||||
|
var stale = row("assistant", "库存 98765", now.minusHours(2));
|
||||||
|
stale.setContentParts("stale structured results");
|
||||||
|
stale.setMetadata("stale tool calls");
|
||||||
|
var hot = row("user", "按仓库分组", now.minusMinutes(3));
|
||||||
|
var answer = row("assistant", "recent answer", now.minusMinutes(2));
|
||||||
|
var result = policy.select(List.of(cold, warm, stale, hot, answer), now);
|
||||||
|
assertEquals(4, result.size());
|
||||||
|
assertTrue(result.getFirst().getContent().contains("华北库存是多少"));
|
||||||
|
assertTrue(result.get(1).getContent().contains("expired"));
|
||||||
|
assertNull(result.get(1).getContentParts());
|
||||||
|
assertNull(result.get(1).getMetadata());
|
||||||
|
assertSame(hot, result.get(2));
|
||||||
|
assertSame(answer, result.get(3));
|
||||||
|
assertEquals("库存 98765", stale.getContent());
|
||||||
|
assertEquals("stale structured results", stale.getContentParts());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void freshSummaryCannotLaunderOldResultsAndUnknownDatesAreNotFresh() {
|
||||||
|
var summary = row("system", "compression with stale totals", now.minusSeconds(1));
|
||||||
|
summary.setMetadata("compression_summary");
|
||||||
|
assertTrue(policy.select(List.of(summary, row("assistant", "unknown age", null)), now).isEmpty());
|
||||||
|
assertTrue(policy.select(List.of(row("user", "future", now.plusHours(1))), now).isEmpty());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void cutoffAgesToolCallAndResponseAsWholeTurn() {
|
||||||
|
var user = row("user", "query", now.minusMinutes(31));
|
||||||
|
var call = row("assistant", "tool call", now.minusMinutes(30));
|
||||||
|
var tool = row("tool", "stale tool payload", now.minusMinutes(29));
|
||||||
|
var answer = row("assistant", "stale paraphrase", now.minusMinutes(28));
|
||||||
|
var result = policy.select(List.of(user, call, tool, answer), now);
|
||||||
|
assertEquals(2, result.size());
|
||||||
|
assertTrue(result.stream().noneMatch(r -> r.getContent().contains("stale")));
|
||||||
|
var fresh = List.of(row("user", "query", now.minusMinutes(30)), call, tool, answer);
|
||||||
|
assertEquals(fresh, policy.select(fresh, now));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void exactWarmBoundaryAndConfigurationAreHonored() {
|
||||||
|
assertEquals(2, policy.select(List.of(row("user", "boundary", now.minusDays(1))), now).size());
|
||||||
|
assertTrue(policy.select(List.of(row("user", "old", now.minusDays(1).minusNanos(1))), now).isEmpty());
|
||||||
|
properties.setHotMinutes(5);
|
||||||
|
properties.setWarmMinutes(10);
|
||||||
|
assertTrue(policy.select(List.of(row("user", "old", now.minusMinutes(11))), now).isEmpty());
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void onlyChannelTurnsUsePolicyAndCanOptOut() {
|
||||||
|
assertFalse(policy.applies(ChatOrigin.EMPTY));
|
||||||
|
assertFalse(policy.applies(ChatOrigin.web("c", "u", 1L, null)));
|
||||||
|
assertFalse(policy.applies(ChatOrigin.cron("c", 1L, null, 2L, null).withSender("u", "feishu", null)));
|
||||||
|
for (String channel : List.of("feishu", "weixin", "wechat", "wecom", "webchat")) {
|
||||||
|
assertTrue(policy.applies(ChatOrigin.EMPTY.withSender("u", channel, null)));
|
||||||
|
}
|
||||||
|
properties.setEnabled(false);
|
||||||
|
assertFalse(policy.applies(ChatOrigin.EMPTY.withSender("u", "feishu", null)));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static MessageEntity row(String role, String content, LocalDateTime created) {
|
||||||
|
var row = new MessageEntity();
|
||||||
|
row.setRole(role);
|
||||||
|
row.setContent(content);
|
||||||
|
row.setCreateTime(created);
|
||||||
|
return row;
|
||||||
|
}
|
||||||
|
}
|
||||||
@ -122,4 +122,40 @@ class DshConversationHistoryTest {
|
|||||||
jdbc.update("INSERT INTO mate_message(id,conversation_id,role,content,status) VALUES(?,?,?,?,?)",
|
jdbc.update("INSERT INTO mate_message(id,conversation_id,role,content,status) VALUES(?,?,?,?,?)",
|
||||||
id, conversation, role, content, status);
|
id, conversation, role, content, status);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void channelHistoryExpiresAnswersButKeepsWarmQuestionAndHotDialogue() {
|
||||||
|
row(1, "a", "user", "cold question", "completed");
|
||||||
|
row(2, "a", "user", "查询华北库存", "completed");
|
||||||
|
row(3, "a", "assistant", "stale inventory 98765", "completed");
|
||||||
|
row(4, "a", "user", "按仓库分组", "completed");
|
||||||
|
row(5, "a", "assistant", "recent answer", "completed");
|
||||||
|
row(6, "a", "user", "current question", "completed");
|
||||||
|
jdbc.update("UPDATE mate_message SET create_time=? WHERE id=1", java.time.LocalDateTime.now().minusDays(2));
|
||||||
|
jdbc.update("UPDATE mate_message SET create_time=? WHERE id IN (2,3)", java.time.LocalDateTime.now().minusHours(2));
|
||||||
|
ChatOrigin channel = ChatOrigin.EMPTY.withSender("u", "weixin", null).withOriginMessageId(6L);
|
||||||
|
String prompt = history.enrich("a", "current question", "current enriched", channel);
|
||||||
|
assertFalse(prompt.contains("cold question"));
|
||||||
|
assertFalse(prompt.contains("98765"));
|
||||||
|
assertFalse(prompt.contains("current question"));
|
||||||
|
assertTrue(prompt.contains("查询华北库存"));
|
||||||
|
assertTrue(prompt.contains("recent answer"));
|
||||||
|
assertTrue(prompt.contains("query the authoritative source in this turn"));
|
||||||
|
assertTrue(prompt.endsWith("current enriched"));
|
||||||
|
assertEquals("stale inventory 98765", jdbc.queryForObject("SELECT content FROM mate_message WHERE id=3", String.class));
|
||||||
|
// Web replay remains compatible, including older text.
|
||||||
|
assertTrue(history.enrich("a", "current question", "current", ChatOrigin.EMPTY).contains("98765"));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void channelWithoutOriginIdDeduplicatesBeforeWarmProjectionAndGuidesEmptyHistory() {
|
||||||
|
row(1, "a", "user", "current", "completed");
|
||||||
|
jdbc.update("UPDATE mate_message SET create_time=?", java.time.LocalDateTime.now().minusHours(2));
|
||||||
|
ChatOrigin channel = ChatOrigin.EMPTY.withSender("u", "feishu", null);
|
||||||
|
String prompt = history.enrich("a", "current", "explicit question", channel);
|
||||||
|
assertFalse(prompt.contains("Historical request"));
|
||||||
|
assertTrue(prompt.contains("freshness"));
|
||||||
|
assertTrue(prompt.endsWith("explicit question"));
|
||||||
|
assertTrue(history.enrich("empty", "query", "query", channel).contains("freshness"));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user