记忆系统 + 多 Agent 协作

日期：2026-06-07
主题：记忆系统 + 多 Agent 协作
进度：32/40 概念

---

一、短期记忆（对话上下文管理）

什么是短期记忆

短期记忆 = 当前对话的上下文。Agent 需要”记住”用户之前说了什么，才能连贯对话。

用户：我想买一台笔记本
Agent：您有什么需求？（记住：用户要买笔记本）
用户：预算 5000 左右
Agent：推荐以下几款 5000 元档的笔记本...（记住：笔记本 + 5000预算）

上下文窗口的矛盾

上下文窗口有限（比如 128K tokens）
+ 对话历史持续增长
= 一定会超出窗口

三种管理策略

① 滑动窗口

只保留最近 N 轮对话，丢弃早期内容。

def sliding_window(history, max_turns=10):
    return history[-max_turns * 2:]  # 每轮有 user + assistant

优点缺点实现最简单丢失早期重要信息Token 消耗可控用户问”我最开始说了什么”就答不上来

② 摘要压缩

定期把旧对话压缩成摘要。

def summarize_history(history, threshold=6000):
    if count_tokens(history) > threshold:
        old_messages = history[:-6]  # 保留最近3轮
        recent_messages = history[-6:]
​
        summary = llm_summarize(old_messages)  # 用 LLM 做摘要
        return [{"role": "system", "content": f"对话摘要：{summary}"}] + recent_messages
    return history

优点缺点保留关键信息摘要过程消耗 Token上下文连贯细节会丢失

③ 混合策略（推荐）

系统提示词（固定）
+ 对话摘要（压缩后的历史）
+ 最近 N 轮原始对话（精确上下文）
+ 用户当前输入

def build_context(system_prompt, history, user_input, max_recent=6):
    tokens_budget = 8000
​
    # 1. 系统提示词（固定开销）
    system_tokens = count_tokens(system_prompt)
    remaining = tokens_budget - system_tokens
​
    # 2. 最近对话（优先保留）
    recent = history[-max_recent:]
    recent_tokens = count_tokens(recent)
    remaining -= recent_tokens
​
    # 3. 旧对话压缩成摘要（用剩余预算）
    if remaining > 500 and len(history) > max_recent:
        old = history[:-max_recent]
        summary = summarize(old, max_tokens=remaining // 2)
        return [system_prompt, summary] + recent + [user_input]
​
    return [system_prompt] + recent + [user_input]

Token 预算分配经验

总窗口 128K tokens 的分配建议：
├── 系统提示词 + 工具描述：2-5K（固定）
├── RAG 检索结果：2-8K（动态）
├── 对话历史：4-20K（动态，需管理）
└── 模型输出预留：2-4K

---

二、长期记忆（向量数据库存储）

什么是长期记忆

长期记忆 = 跨对话持久化的信息。用户下次来，Agent 还记得。

对话1：用户说"我喜欢简洁风格的UI"
对话2：用户说"帮我设计一个页面"
Agent：根据你之前提到的偏好，我设计了一个简洁风格的页面...

长期记忆的存储结构

memory_entry = {
    "id": "mem_001",
    "content": "用户偏好简洁风格的UI设计",
    "type": "preference",        # 记忆类型
    "user_id": "user_123",       # 所属用户
    "created_at": "2026-06-07",  # 创建时间
    "last_accessed": "2026-06-07",  # 最后访问时间
    "importance": 0.8,           # 重要性评分
    "embedding": [0.1, 0.3, ...] # 向量
}

记忆的 CRUD 操作

写入记忆：

def save_memory(content, user_id, memory_type):
    embedding = get_embedding(content)
    memory_store.add({
        "content": content,
        "user_id": user_id,
        "type": memory_type,
        "embedding": embedding,
        "importance": calculate_importance(content)
    })

检索记忆：

def recall_memories(query, user_id, top_k=5):
    query_embedding = get_embedding(query)
    results = memory_store.query(
        embedding=query_embedding,
        filter={"user_id": user_id},
        top_k=top_k
    )
    return results

更新记忆：

def update_memory(memory_id, new_content):
    # 合并旧记忆和新信息
    old = memory_store.get(memory_id)
    merged = llm_merge(old.content, new_content)
    memory_store.update(memory_id, {
        "content": merged,
        "embedding": get_embedding(merged)
    })

记忆的生命周期

新记忆 → 短期缓存 → 重要则持久化 → 定期清理过期/低频记忆
                                      ↓
                                重要性衰减 → 不再访问 → 归档/删除

重要性评分规则：

信号加分用户明确说”记住这个”+0.5被多次访问+0.1/次包含偏好/习惯类信息+0.3最近创建+0.2

记忆去重与合并

用户可能在不同对话中表达相似的信息：

对话1："我喜欢Python"
对话3："我平时主要用Python写后端"
→ 应该合并成一条记忆，而不是两条

def add_memory_with_dedup(new_content, user_id):
    # 先检索相似记忆
    similar = recall_memories(new_content, user_id, top_k=3)
    if similar and cosine_similarity(new_content, similar[0]) > 0.85:
        # 合并到已有记忆
        update_memory(similar[0].id, new_content)
    else:
        # 作为新记忆保存
        save_memory(new_content, user_id)

---

三、工作记忆（任务中间状态）

什么是工作记忆

工作记忆 = 当前任务的中间结果和状态。类似草稿纸。

任务：分析销售数据并生成报告

工作记忆：
├── 已加载数据：sales_2026.csv（10000行）
├── 分析结果：Q1下降15%，Q2回升8%
├── 已生成图表：趋势图、分布图
└── 待完成：汇总报告、添加结论

Scratchpad 模式

class Scratchpad:
    """Agent 的草稿纸"""

    def __init__(self):
        self.notes = []
        self.results = {}
        self.plan = []
        self.current_step = 0

    def note(self, content):
        """记录观察和想法"""
        self.notes.append({"time": now(), "content": content})

    def save_result(self, key, value):
        """保存中间结果"""
        self.results[key] = value

    def get_context(self):
        """生成工作记忆上下文"""
        return f"""
当前任务进度：
计划：{self.plan}
当前步骤：{self.current_step}/{len(self.plan)}
已完成的结果：{self.results}
笔记：{self.notes[-5:]}  # 最近5条
"""

工作记忆 vs 对话历史

维度对话历史工作记忆内容用户和Agent的对话任务中间状态生命周期跨任务仅当前任务格式消息列表结构化状态用途上下文连贯任务追踪

工作记忆的注入方式

def build_messages(user_input, chat_history, scratchpad):
    messages = []

    # 系统提示词
    messages.append({"role": "system", "content": system_prompt})

    # 注入工作记忆（让 LLM 知道当前进度）
    if scratchpad.is_active():
        messages.append({
            "role": "system",
            "content": f"当前任务状态：\n{scratchpad.get_context()}"
        })

    # 对话历史
    messages.extend(chat_history)

    # 用户输入
    messages.append({"role": "user", "content": user_input})

    return messages

---

四、记忆检索与摘要压缩

何时压缩

信号行动Token 使用 > 80% 窗口压缩旧对话对话轮次 > 20 轮压缩前 15 轮用户切换话题压缩旧话题的详细内容

如何检索相关记忆

def get_relevant_context(user_input, user_id, chat_history):
    context_parts = []

    # 1. 短期记忆：最近对话
    recent = chat_history[-6:]
    context_parts.append(("recent_chat", recent))

    # 2. 长期记忆：语义相关
    long_term = recall_memories(user_input, user_id, top_k=3)
    context_parts.append(("long_term", long_term))

    # 3. 工作记忆：当前任务状态
    if scratchpad.is_active():
        context_parts.append(("task_state", scratchpad.get_context()))

    return context_parts

摘要压缩的 Prompt

SUMMARIZE_PROMPT = """
请将以下对话历史压缩为摘要。要求：
1. 保留关键事实和决策
2. 保留用户表达的偏好和需求
3. 保留未完成的任务
4. 删除寒暄、重复、确认类对话
5. 控制在200字以内

对话历史：
{history}

摘要：
"""

三种记忆的协作

用户输入
  ↓
┌─────────────────────────────────────────┐
│ 检索相关记忆                              │
│  ├── 短期：最近 3 轮对话                  │
│  ├── 长期：语义相关的用户偏好              │
│  └── 工作：当前任务进度                    │
└─────────────────────────────────────────┘
  ↓
组装成完整上下文 → 发给 LLM
  ↓
LLM 回复
  ↓
更新记忆
  ├── 短期：追加到对话历史
  ├── 长期：提取新偏好/事实保存
  └── 工作：更新任务进度

---

五、Agent 角色定义与分工

为什么需要多 Agent

单个 Agent 的局限：

• 系统提示词太长，角色混乱
• 一个 LLM 难以同时擅长多种任务
• 复杂任务需要不同”专业”的 Agent 协作

角色设计原则

① 单一职责

❌ 万能Agent："你既能写代码，又能写文案，还能做数据分析"
✅ 专精Agent：
   - CoderAgent：只负责写代码
   - WriterAgent：只负责写文案
   - AnalystAgent：只负责数据分析

② 能力边界清晰

coder_agent = {
    "role": "代码工程师",
    "capabilities": ["写代码", "调试", "代码审查"],
    "cannot_do": ["写文案", "做设计", "管理项目"],
    "tools": ["code_executor", "file_reader", "git"]
}

③ 输入输出明确

WriterAgent：
  输入：{ "topic": "...", "style": "...", "length": "..." }
  输出：{ "content": "...", "word_count": 1234 }

常见角色模式

模式角色适用场景研究 + 写作Researcher + Writer报告生成规划 + 执行Planner + Executor复杂任务主代码 + 审查Coder + Reviewer代码开发竞品分析多个 Researcher + Analyst市场调研

---

六、通信协议与消息传递

消息格式

message = {
    "id": "msg_001",
    "from": "researcher",
    "to": "writer",
    "type": "task_result",      # 消息类型
    "content": "调研结果：...",
    "metadata": {
        "task_id": "task_123",
        "status": "completed"
    },
    "timestamp": "2026-06-07T10:30:00Z"
}

消息类型

类型用途示例task_assign分配任务Supervisor → Workertask_result返回结果Worker → Supervisorquestion请求帮助Worker → 另一个 Workeranswer回答问题另一个 Worker → Workerfeedback评价反馈Reviewer → Coder

通信模式

① 直接通信

Agent A 直接发消息给 Agent B。

class Agent:
    def send(self, to_agent, message):
        to_agent.receive(message)

    def receive(self, message):
        self.message_queue.append(message)

② 通过总线通信

所有 Agent 通过中央消息总线通信。

class MessageBus:
    def __init__(self):
        self.subscribers = defaultdict(list)

    def subscribe(self, agent, message_type):
        self.subscribers[message_type].append(agent)

    def publish(self, message):
        for agent in self.subscribers[message["type"]]:
            agent.receive(message)

③ 异步通信

async def collaborate(task):
    # 并行发送任务
    results = await asyncio.gather(
        agent_a.execute(subtask_1),
        agent_b.execute(subtask_2),
        agent_c.execute(subtask_3)
    )
    # 汇总结果
    return merge_results(results)

---

七、监督者模式 vs 对等协作模式

监督者模式（Supervisor）

一个”老板”Agent 负责分配任务、汇总结果、做最终决策。

         ┌──────────┐
         │ Supervisor│ ← 用户直接交互
         └────┬─────┘
      ┌───────┼───────┐
      ↓       ↓       ↓
  ┌──────┐ ┌──────┐ ┌──────┐
  │Worker│ │Worker│ │Worker│
  │  A   │ │  B   │ │  C   │
  └──────┘ └──────┘ └──────┘

代码结构：

class Supervisor:
    def __init__(self, workers):
        self.workers = workers
        self.llm = LLM()

    def handle(self, user_request):
        # 1. 规划：决定用哪些 worker
        plan = self.llm.plan(user_request, self.workers)

        # 2. 分配：逐个或并行分配子任务
        results = []
        for step in plan:
            worker = self.workers[step.worker_name]
            result = worker.execute(step.task)
            results.append(result)

        # 3. 汇总：整合所有结果
        final = self.llm.summarize(results)
        return final

优点：可控、可追踪、容易调试
缺点：Supervisor 是瓶颈、通信开销大

对等协作模式（Peer-to-Peer）

Agent 之间平等对话，没有”老板”。

  ┌──────┐    ┌──────┐
  │Agent │←──→│Agent │
  │  A   │    │  B   │
  └──┬───┘    └───┬───┘
     │            │
     └──→┌──────┐←┘
         │Agent │
         │  C   │
         └──────┘

代码结构：

class PeerAgent:
    def __init__(self, name, peers):
        self.name = name
        self.peers = peers  # 其他 Agent 的引用

    def collaborate(self, task):
        # 自己能做的自己做
        my_result = self.execute(task)

        # 需要帮助时找合适的 peer
        if needs_help(my_result):
            helper = self.find_best_peer(task)
            help_result = helper.collaborate(subtask)
            return merge(my_result, help_result)

        return my_result

优点：灵活、无单点瓶颈
缺点：难调试、可能死循环、冲突多

选择指南

维度监督者模式对等协作模式适用任务流程明确、可预先规划开放式、需要创意碰撞可控性高低调试难度低高扩展性受限于 Supervisor更好典型场景流水线、报告生成研讨、头脑风暴

---

八、冲突解决与共识机制

冲突场景

Agent A："应该用 React 开发"
Agent B："应该用 Vue 开发"
→ 冲突！谁来决定？

三种解决策略

① Supervisor 裁决

class SupervisorArbiter:
    def resolve(self, conflict):
        # 收集各方意见
        opinions = conflict.opinions

        # Supervisor 综合评估
        decision = self.llm.evaluate(
            f"关于'{conflict.topic}'，有以下意见：\n"
            f"{format_opinions(opinions)}\n"
            f"请评估各方理由，给出最终决策。"
        )
        return decision

② 投票机制

def majority_vote(agents, question):
    votes = {}
    for agent in agents:
        answer = agent.answer(question)
        votes[answer] = votes.get(answer, 0) + 1

    return max(votes, key=votes.get)

③ 辩论机制

def debate(agents, topic, rounds=3):
    arguments = []
    for round in range(rounds):
        for agent in agents:
            arg = agent.argue(topic, previous_args=arguments)
            arguments.append(arg)

    # 最后由裁判 Agent 做决策
    judge = JudgeAgent()
    return judge.decide(topic, arguments)

选择策略

策略适用场景开销Supervisor 裁决决策影响大、需要权威低投票多个 Agent 意见权重相等低辩论复杂问题需要深入讨论高

---

九、Day 5 核心认知总结

1. 短期记忆 = 滑动窗口 + 摘要压缩：保留最近对话，压缩旧对话
2. 长期记忆 = 向量存储 + 语义检索：跨对话记住用户偏好
3. 工作记忆 = Scratchpad：当前任务的中间状态
4. 三种记忆协作：短期保连贯、长期保持久、工作保进度
5. 角色设计要专精：单一职责、能力边界清晰
6. 监督者模式适合流程明确的任务，对等模式适合开放探索
7. 冲突解决三种策略：裁决、投票、辩论，根据场景选择

‍