Gliding Horse 上下文动态感知与智能压缩：用 RelevanceTracker 让 Agent 真正“听得进”每一句话

1. 长对话里 Agent 为什么突然“听不见”你说话如果你正在做 Gliding Horse 这类带多轮记忆的 Agent大概率遇到过这种场景前几轮聊得好好的用户在第 12 轮补了一句“刚才那个接口地址改成测试环境”Agent 却像没听见继续拿生产地址往下跑。你翻日志发现消息其实收到了但要么被塞进一个已经失效的注入方法里要么在上下文淘汰时被当成“低价值历史”清掉了。这不是模型笨是上下文管理没做好动态感知。长对话里 Token 是有限资源传统做法要么全量保留导致上下文膨胀、推理变慢变贵要么粗暴截断末尾几条把用户刚补充的关键约束一起扔掉。Gliding Horse 的 RelevanceTracker 就是来解决这个问题的它给每条输入实时打一个任务关联度分数 relevance_score让淘汰和压缩策略知道“哪句话该记牢哪句话可以早点忘”。这篇我会从原问题出发给出可复制的 RelevanceTracker 配置骨架、压缩阈值参数再演示多轮对话下的验证动作。适合正在自建 Agent、被上下文膨胀和补充输入丢失折磨的开发者。全程用 Gliding Horse 的组件名和事件流来讲你可以直接对照自己的代码改。2. 前置准备TaoToken 接入与 Embedding 管线打通RelevanceTracker 的核心是算余弦相似度所以你得先有一条能用的 Embedding 管线。Gliding Horse 支持 Ollama / OneAPI 等后端我这边统一走 TaoToken 的 API 来拿模型能力省得本地再维护一套向量服务。先到官网 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 了解整体能力然后进控制台创建 API Key。地址是 https://taotoken.net/api 注意这个不带 UTM直接配到环境变量里就行。export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/api如果你还没建 Key直接去 https://taotoken.net/api-keys 生成权限选最小可用即可。接入文档在 https://taotoken.net/doc 里面有 Embedding 和对话模型的调用示例建议先跑通一次 embedding 请求再往下走。注意RelevanceTracker 每次用户输入都会调两次 embedding当前输入 任务描述如果任务描述很长建议缓存 task_vec不要每轮重算否则延迟会明显上升。3. 可复制配置RelevanceTracker 骨架与压缩阈值参数3.1 RelevanceTracker 评分器核心逻辑就两个相似度加权全局任务相关度和任务 5W2H 描述的余弦相似度占大头局部连贯性和上一条输入的相似度占小头。α 默认 0.6。import numpy as np from dataclasses import dataclass from typing import List def cosine_similarity(a: List[float], b: List[float]) - float: va, vb np.array(a), np.array(b) denom (np.linalg.norm(va) * np.linalg.norm(vb)) if denom 0: return 0.0 return float(np.dot(va, vb) / denom) class RelevanceTracker: def __init__(self, embedding_service, alpha: float 0.6): self.embedding embedding_service self.alpha alpha self._task_vec_cache {} def _task_vec(self, task_description: str): if task_description not in self._task_vec_cache: self._task_vec_cache[task_description] self.embedding.embed(task_description) return self._task_vec_cache[task_description] def compute_relevance(self, user_input: str, task_description: str, prev_input: str) - float: input_vec self.embedding.embed(user_input) task_vec self._task_vec(task_description) global_sim cosine_similarity(input_vec, task_vec) prev_vec self.embedding.embed(prev_input) if prev_input else input_vec local_sim cosine_similarity(input_vec, prev_vec) score self.alpha * global_sim (1 - self.alpha) * local_sim return round(score, 4)3.2 淘汰与压缩阈值参数L1Session 的 EvictionConfig 我建议这样起步后面按场景微调参数默认值作用调大后果relevance_threshold0.3硬淘汰阈值低于它且超安全窗口直接丢淘汰更激进可能误伤safe_window_seconds300安全窗口窗口内不硬淘汰上下文保留更久Token 涨beta0.5融合评分里当前查询相似度 vs 任务历史关联度的权重更偏向当前查询alpha0.6全局任务相关度权重更看重任务目标而非连贯topic_shift_threshold0.4相邻 turns 相似度低于它视为话题切换更容易触发主动压缩3.3 补充输入暂存与注入SupplementaryInputStore 是修“补充输入丢失”的关键线程安全的内存共享存储SA 分类后存AgentRunner 在 CycleStart 消费。import threading from dataclasses import dataclass from typing import Dict, List dataclass class SupplementEntry: content: str embedding: List[float] relevance_score: float class SupplementaryInputStore: def __init__(self): self._store: Dict[str, List[SupplementEntry]] {} self._lock threading.Lock() def store(self, task_iri: str, entry: SupplementEntry): with self._lock: self._store.setdefault(task_iri, []).append(entry) def consume(self, task_iri: str) - List[SupplementEntry]: with self._lock: return self._store.pop(task_iri, [])AgentRunner 在 CycleStart 里这样接entries supplementary_store.consume(task_iri) for e in entries: l1_session.append_turn( contente.content, relevance_scoree.relevance_score, is_supplementTrue, ) messages.append({role: user, content: e.content})is_supplementTrue的条目在 evict_with_query 里不受硬阈值淘汰这是保证用户纠正不被误删的关键。4. 验证请求多轮对话下确认 Agent 真的“听得进”配置完别急着上生产先用一段多话题长会话验证。我构造了 15 轮对话第 3 轮给任务目标第 8 轮插入一个话题漂移第 12 轮补充关键约束看 Agent 是否还能抓住第 12 轮的内容。tracker RelevanceTracker(embedding_servicetaotoken_embedder, alpha0.6) task_desc 把订单服务从生产库迁移到测试库只改连接串不动业务逻辑 turns [ 帮我规划一下订单服务的迁移步骤, 先看现在用的连接串配置在哪, 目标是测试库别碰生产数据, # ... 中间省略若干轮闲聊 顺便问下今天天气怎么样, 刚才说的连接串改成 jdbc:mysql://test-db:3306/order, ] prev for i, t in enumerate(turns): score tracker.compute_relevance(t, task_desc, prev) print(fturn {i}: score{score} | {t[:20]}) prev t实测下来第 12 轮那句“改成 test-db”的 score 会明显高于“今天天气”那轮因为全局任务相关度把它拉起来了。你可以在日志里看到类似turn 3: score0.8123 | 目标是测试库别碰生产数据 turn 11: score0.2145 | 顺便问下今天天气怎么样 turn 12: score0.7641 | 刚才说的连接串改成 jdbc:mysql://test-db然后检查 L1 窗口天气那轮如果超过 safe_window_seconds 且 score 低于 0.3会被硬淘汰第 12 轮因为 is_supplementTrue即使分数波动也不会被硬阈值清掉。再触发一次 topic_coherence_agent 的全窗口重算确认话题切换事件 TopicShiftDetected 被正确发布主动压缩后旧话题残留被清理。如果你想让验证更直观可以到 https://taotoken.net/models 用模型对话直接对比压缩前后的回答差异看 Agent 是否还引用第 12 轮的约束。5. 本篇常见错排查补充输入还是丢先确认 SA 分类后走的是 SupplementaryInputStore.store而不是旧的注入方法。再看 AgentRunner 的 CycleStart 有没有调 consume很多人只存不取。relevance_score 全是 0 或 NaN多半是 embedding 返回了空向量或维度不一致。检查 TaoToken 的 embedding 接口返回结构确认 cosine_similarity 里 denom 不为 0。淘汰太激进关键历史被删把 relevance_threshold 从 0.3 降到 0.2或把 safe_window_seconds 从 300 提到 600。也可以调大 beta让任务历史关联度在融合评分里权重更高。话题检测频繁误触发topic_shift_threshold 默认 0.4 偏敏感短对话里相邻 turns 相似度本来就低。建议提到 0.5或把检测周期从 2 分钟放宽到 5 分钟。Token 没降下来确认 ContextWindowManager 压缩时真的读了 relevance_score 映射。如果还是机械截末尾压缩结果不会聚焦Token 自然降不下去。embedding 延迟高task_vec 一定要缓存别每轮重算。补充输入的 embedding 可以在 store 时就生成好消费时直接用。6. 下一步把感知能力接进你的 Agent 循环这套机制跑通后你可以按自己的场景调 α、β、硬阈值和安全窗口。长期做编码类 Agent 或需要跨会话记忆的场景建议把 Coding Plan 用起来配合 RelevanceTracker 做更细粒度的上下文预算控制入口在 https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content 。接入过程中如果遇到 embedding 或淘汰策略的问题直接翻接入文档 https://taotoken.net/doc 里面的事件总线和 Hook 说明能帮你少踩不少坑。我自己的经验是先把 SupplementaryInputStore 和 is_supplement 保护做扎实再调阈值顺序反了会一直觉得“怎么调都不对”。