3.2. 参考资源与记忆管理是 Skills 保持状态的基石

参考资源与记忆管理是 Skills 保持状态的基石

本章定位:实战教程
预计完成时间:约 60 分钟
核心问题:如何让 Skill 记住上下文、引用外部知识、并在多次调用间保持状态?


想象这样一个场景:你花了两小时教会一个客服 Skill 你公司的产品线、退货政策和常见异常处理流程。第一次对话效果惊艳。三分钟后,同一个用户再次提问,Skill 却像从未见过你一样,从头开始解释退货流程。这不是推理能力的问题——Skill 失忆了。

没有记忆的 Skill 就像每次醒来都失忆的专家。推理能力再强,也无法积累经验、引用历史、感知上下文。参考资源与记忆管理,正是 Skills 从“一次性推理引擎”进化为“上下文感知系统”的基石。

本章将带你构建三层记忆体系:文件型参考资源(长期记忆)、会话历史存储(短期记忆)、向量语义检索(联想记忆),让你的 Skill 真正拥有“记住”的能力。


一、你需要什么

组件 用途 要求
Python 环境 运行 Skill 和记忆管理代码 3.10+
Redis 或 PostgreSQL 存储会话历史 Redis 7.x 或 PG 15+
向量数据库 语义检索 Qdrant / Milvus / ChromaDB 任选其一
Embedding 模型 文本转向量 OpenAI text-embedding-3-small 或本地 bge-large-zh
文本编辑器 编写 SKILL.md 和参考文件 VS Code 或同类工具

注意:如果你尚未搭建向量数据库,推荐从 ChromaDB 开始——纯 Python 实现,零配置启动,适合快速验证。生产环境建议迁移到 Qdrant 或 Milvus。


二、最终成果

完成本章后,你将拥有一个具备三层记忆的 Skill 系统:

┌─────────────────────────────────────────────┐
│                  Skill 记忆层               │
├───────────────┬─────────────┬───────────────┤
│  参考资源层   │  会话记忆层  │  向量检索层   │
│ (references/) │ (Redis/PG)  │  (Qdrant等)   │
├───────────────┼─────────────┼───────────────┤
│ 领域知识文档  │ 对话历史    │ 语义相似搜索   │
│ 政策/FAQ/模板 │ 窗口管理    │ 跨会话联想    │
│ 版本化存储    │ 自动摘要    │ 时效性控制    │
└───────────────┴─────────────┴───────────────┘

为什么做这个? Skill 的“上下文感知”不是魔法,而是分层记忆设计的结果。参考资源提供领域知识的确定性基础,会话记忆维持对话连贯性,向量检索赋予跨会话的联想能力。三者组合,才能让 Skill 在多次交互中持续积累状态。


三、步骤说明

步骤一:文件型参考资源的版本化与检索

文件型参考资源是最基础、最可控的记忆形式——将领域知识以 Markdown 或 YAML 文件存储在 references/ 目录下,Skill 按需加载。

1.1 创建目录结构

# 在你的 Skill 项目根目录下执行
mkdir -p my-skill/references/policies
mkdir -p my-skill/references/faq
mkdir -p my-skill/references/templates

1.2 编写参考文档与索引

references/ 下创建一个总索引文件,帮助 Agent 快速定位所需文档:

# references/index.md

## 文档索引
- [退货政策](policies/refund-policy.md) - 标准退货流程、时限、例外条款
- [定价FAQ](faq/pricing-faq.md) - 常见定价问题与报价模板
- [异常处理SOP](templates/exception-sop.md) - 升级处理的标准操作流程

关键设计原则:索引文件不是给人类看的目录,而是 Agent 在“发现阶段”判断是否加载某个参考文件的决策依据。每个条目的描述应足够精确,让模型仅凭描述就能判断相关性。

1.3 在 SKILL.md 中声明参考资源

按照 Agent Skills 规范的渐进式披露模式,仅在 SKILL.md 中声明参考资源的存在,不预先加载内容:

# SKILL.md(节选)

## 参考资源
当需要查询领域政策、FAQ 或操作模板时,请查阅 `references/` 目录:
- 先读取 `references/index.md` 定位相关文件
- 再根据索引加载具体文档内容
- 参考文件为版本化存储,以目录名中的日期为准(如 `policies-2025Q4/`)

预期结果

Agent 在对话中首次遇到退货相关问题时,会经历以下流程:

  1. 读取 references/index.md,识别“退货政策”条目
  2. 加载 references/policies/refund-policy.md 获取完整条款
  3. 基于文档内容生成回复

之后同一对话中再次涉及退货问题时,Agent 可直接调用已加载的记忆,无需重复读取。

踩坑经验references/ 的加载效果依赖 Agent 的具体实现。如果 SKILL.md 中未明确指示“何时查阅参考文件”,Agent 可能直接依赖训练数据回复,完全忽略参考资源。务必在 SKILL.md 的触发条件中写明:“遇到 [X] 类问题时,必须先查阅 references/ 下的 [Y] 文件”。


步骤二:会话记忆的存储与窗口管理

参考资源解决了领域知识的长期记忆问题,但对话本身也需要被记住——用户上一轮说了什么、Skill 上一轮做了什么决策,这些信息必须在同一会话中保持连贯。

2.1 设计会话记忆存储结构

使用 Redis 存储对话历史,每条消息包含角色、内容和时间戳:

# memory/session_store.py
import redis
import json
from datetime import datetime
from typing import List, Dict

class SessionMemory:
    def __init__(self, redis_url: str = "redis://localhost:6379"):
        self.client = redis.from_url(redis_url)
        self.max_history = 20  # 最大保留消息数

    def append(self, session_id: str, role: str, content: str):
        """向会话追加一条消息"""
        key = f"session:{session_id}:history"
        message = {
            "role": role,
            "content": content,
            "timestamp": datetime.now().isoformat()
        }
        self.client.rpush(key, json.dumps(message))
        # 控制窗口大小:超出上限时从左侧裁剪
        current_len = self.client.llen(key)
        if current_len > self.max_history:
            self.client.ltrim(key, current_len - self.max_history, -1)

    def get_history(self, session_id: str, n: int = None) -> List[Dict]:
        """获取会话历史,n=None 表示获取全部"""
        key = f"session:{session_id}:history"
        end = -1
        start = -n if n else 0
        raw = self.client.lrange(key, start, end)
        return [json.loads(m) for m in raw]

    def summarize_and_compress(self, session_id: str, llm_client):
        """对早期消息进行摘要,压缩上下文窗口"""
        key = f"session:{session_id}:history"
        messages = self.get_history(session_id)
        if len(messages) <= 10:
            return  # 消息较少,无需压缩

        # 将前50%的消息送去摘要
        split_point = len(messages) // 2
        older = messages[:split_point]

        summary = llm_client.generate(
            f"用一段话总结以下对话的关键信息:{json.dumps(older)}"
        )

        # 用摘要替换旧消息
        self.client.ltrim(key, split_point, -1)  # 删除前半段
        self.client.lpush(key, json.dumps({      # 将摘要插入最前
            "role": "system",
            "content": f"[对话摘要] {summary}",
            "timestamp": datetime.now().isoformat()
        }))

2.2 集成到 Skill 调用流程

# skill_runner.py
from memory.session_store import SessionMemory

memory = SessionMemory()

def run_skill(session_id: str, user_input: str, skill_func):
    # 1. 加载会话历史
    history = memory.get_history(session_id, n=10)

    # 2. 将历史注入 Skill 的上下文
    context_with_history = {
        "current_input": user_input,
        "conversation_history": history
    }

    # 3. 执行 Skill
    response = skill_func(context_with_history)

    # 4. 保存用户输入和 Skill 回复
    memory.append(session_id, "user", user_input)
    memory.append(session_id, "assistant", response)

    # 5. 必要时压缩历史
    memory.summarize_and_compress(session_id, llm_client)

    return response

预期结果

  • 同一 session_id 下的多轮对话,Skill 能记住前文提到的用户偏好、已确认的信息和中间决策
  • 超过 20 条消息时自动裁剪最早的消息,防止上下文窗口溢出
  • 长对话自动触发摘要压缩,以少量 token 保留早期对话的要点

踩坑经验:上下文窗口并非越大越好。当前调研资料显示,主流模型的指令遵循能力在超过 8K token 后会逐渐下降。保持会话历史控制在 4K token 以内(约 10-15 轮对话),是稳定性与记忆完整性的平衡点。不要试图记忆一切,要学会总结与遗忘。


步骤三:嵌入与向量搜索在 Skills 中的应用

文件参考资源解决“确定性问题”(如政策条款),会话记忆解决“连贯性问题”(如对话上下文),但还有一类“联想性问题”——用户问“之前那个蓝色包装的产品怎么退货”,Skill 需要从历史上百次对话中找出“蓝色包装产品”是哪一次讨论的。

这就是向量语义检索的用武之地。

3.1 初始化向量数据库

# memory/vector_store.py
import chromadb
from chromadb.utils import embedding_functions

class VectorMemory:
    def __init__(self, persist_dir: str = "./chroma_db"):
        self.client = chromadb.PersistentClient(path=persist_dir)

        # 使用 OpenAI 嵌入模型,也可替换为本地模型
        self.embed_fn = embedding_functions.OpenAIEmbeddingFunction(
            api_key="your-api-key",
            model_name="text-embedding-3-small"
        )

        self.collection = self.client.get_or_create_collection(
            name="skill_memories",
            embedding_function=self.embed_fn,
            metadata={"hnsw:space": "cosine"}
        )

    def store(self, doc_id: str, content: str, metadata: dict = None):
        """将文本存入向量库,附带时间戳等元数据"""
        self.collection.add(
            documents=[content],
            ids=[doc_id],
            metadatas=[{
                **(metadata or {}),
                "stored_at": datetime.now().isoformat()
            }]
        )

    def search(
        self, 
        query: str, 
        n_results: int = 5,
        similarity_threshold: float = 0.75
    ) -> list:
        """语义搜索,过滤低相似度结果"""
        results = self.collection.query(
            query_texts=[query],
            n_results=n_results
        )

        # 解析并过滤结果
        filtered = []
        for i, distance in enumerate(results["distances"][0]):
            if distance >= similarity_threshold:  # cosine 距离越大越相似
                filtered.append({
                    "content": results["documents"][0][i],
                    "metadata": results["metadatas"][0][i],
                    "score": distance
                })

        # 按时间戳降序排列,优先返回较新的记忆
        filtered.sort(
            key=lambda x: x["metadata"].get("stored_at", ""),
            reverse=True
        )

        return filtered

3.2 设置相似度阈值与时效性控制

参数 推荐值 说明
similarity_threshold 0.75 低于此值的搜索结果视为不相关,直接丢弃
n_results 5 每次检索返回的最多候选数
时效性衰减 30 天 超过 30 天的记忆降低权重(通过元数据过滤实现)
def search_with_decay(self, query: str, decay_days: int = 30):
    """搜索结果中过滤过期记忆"""
    results = self.search(query)
    cutoff = datetime.now() - timedelta(days=decay_days)

    fresh_results = []
    for r in results:
        stored_at = datetime.fromisoformat(r["metadata"]["stored_at"])
        if stored_at > cutoff:
            fresh_results.append(r)

    return fresh_results

3.3 嵌入向量搜索在 Skill 中的实际应用

场景:客服 Skill 的跨会话联想

# 用户问:"上次说的那个蓝色包装的洗面奶,退货需要原包装吗?"
def handle_user_query(session_id: str, user_input: str):
    # 1. 先在向量库中搜索"蓝色包装洗面奶"相关历史
    vector_memory = VectorMemory()
    related_memories = vector_memory.search(
        query="蓝色包装 洗面奶 退货",
        n_results=3
    )

    # 2. 从会话记忆中加载当前对话历史
    session_memory = SessionMemory()
    current_history = session_memory.get_history(session_id, n=5)

    # 3. 组合记忆,构建完整上下文
    context = {
        "current_query": user_input,
        "current_conversation": current_history,
        "related_past_discussions": [
            m["content"] for m in related_memories
        ],
        "knowledge_base": load_references(["refund-policy.md"])
    }

    # 4. 调用 Skill 生成回复
    return skill_infer(context)

预期结果

  • 向量库中存储了历史对话的关键片段(如“用户购买了蓝色包装的XX洗面奶”)
  • 当用户用不同措辞提及同一概念时,语义搜索能跨越关键词差异找到关联记忆
  • 老旧的记忆(30 天以上)自动衰减,避免噪音干扰

踩坑经验:相似度阈值设置过低会导致大量无关记忆污染上下文,设置过高则会漏掉真正相关的信息。建议从 0.75 开始,在实际业务数据上做 A/B 测试,观察检索结果的相关性分布。另外,嵌入向量的更新频率需要权衡——每次对话都重新嵌入成本高,但过时的嵌入会降低检索质量。折中方案是增量更新 + 定期全量重建。


四、回顾与下一章

你完成了什么

记忆层 技术方案 解决什么问题 核心参数
文件参考资源 references/ 目录 + 索引 领域知识的确定性记忆 索引精度、文件粒度
会话记忆 Redis 队列 + 窗口裁剪 对话连贯性 窗口大小=20 条,目标≤4K token
向量语义检索 ChromaDB + Embedding 跨会话联想 阈值=0.75,衰减=30天

行动清单

  1. 整理你的领域知识,拆分为独立的 references/ 文件,创建索引
  2. 实现会话记忆存储,设置合理的窗口大小和自动摘要触发条件
  3. 部署向量数据库,将关键对话片段和历史决策存储为向量
  4. 设置相似度阈值和时效性衰减参数,在生产数据上验证
  5. 将三层记忆集成到 Skill 调用流程的上下文构建阶段

衔接下一章

三层记忆让 Skill 具备了“状态保持”能力——它能记住过去的对话、引用稳定的知识、联想相关的历史。但真正的专家不仅要有记忆,还要有规划能力:面对“帮我整理这个季度的销售报告”这类复杂请求时,Skill 需要自动拆解步骤、协调多个子任务。

下一章《复杂任务分解是现代 Agent 的必备能力》将带你构建具备规划能力的 Skills,让它们在记忆的基础上,学会制定和执行多步骤计划。

本文章首发在 LearnKu.com 网站上。

上一篇 下一篇
讨论数量: 0
发起讨论 查看所有版本


暂无话题~