主题
原文：https://www.mstx.cn/harness/432.html · 创建：2026-09-06 · 更新：2026-09-06 · 阅读：1082
本节目标
前四节分别讨论了短期记忆、长期记忆、会话状态和压缩策略。这些组件各自解决特定的问题，但真正的价值在于它们组合在一起时产生的效果。
本节将构建一个完整的、带持久记忆的Agent。这个Agent能够：
- 在当前会话中保持对话连续性（短期记忆）
- 跨会话记住用户偏好和事实（长期记忆）
- 跟踪任务的执行状态（会话状态）
- 在上下文窗口压力下智能压缩历史（压缩策略）
场景设定
构建一个个人助理Agent，帮助用户管理日常事务：查询日程、创建任务、记录偏好、搜索信息。
这个Agent的核心特点是：它会记住用户的偏好和习惯，并且随着使用次数的增加，它变得越来越"懂"用户。
第一步：整合记忆系统
将前面各节构建的组件整合为一个统一的记忆系统：
python
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
import json
import os
import uuid
import re
class MessageImportance:
CRITICAL = 5
HIGH = 4
MEDIUM = 3
LOW = 2
MINIMAL = 1
class ManagedMessage:
def __init__(self, role: str, content: str,
importance: int = MessageImportance.MEDIUM,
tool_call_id: str = None,
token_estimate: int = 0,
timestamp: float = 0.0):
self.role = role
self.content = content
self.importance = importance
self.tool_call_id = tool_call_id
self.token_estimate = token_estimate or max(1, len(content) // 2)
self.timestamp = timestamp or datetime.now().timestamp()
def to_api_format(self) -> Dict[str, Any]:
msg = {"role": self.role, "content": self.content}
if self.tool_call_id:
msg["tool_call_id"] = self.tool_call_id
return msg
class WorkingMemory:
"""短期记忆"""
def __init__(self, max_tokens: int = 8000):
self.max_tokens = max_tokens
self.messages: List[ManagedMessage] = []
self.system_prompt: Optional[ManagedMessage] = None
def set_system_prompt(self, content: str):
self.system_prompt = ManagedMessage(
role="system", content=content,
importance=MessageImportance.CRITICAL,
)
def add_message(self, role: str, content: str,
importance: int = MessageImportance.MEDIUM,
tool_call_id: str = None):
msg = ManagedMessage(
role=role, content=content,
importance=importance, tool_call_id=tool_call_id,
)
self.messages.append(msg)
return msg
def get_context(self) -> List[Dict[str, Any]]:
context = []
current_tokens = 0
if self.system_prompt:
context.append(self.system_prompt.to_api_format())
current_tokens += self.system_prompt.token_estimate
reserved = 1500 # 预留输出空间
available = self.max_tokens - current_tokens - reserved
selected = self._select_within_budget(available)
context.extend(selected)
return context
def _select_within_budget(self, budget: int) -> List[Dict[str, Any]]:
total = sum(m.token_estimate for m in self.messages)
if total <= budget:
return [m.to_api_format() for m in self.messages]
# 需要丢弃低重要性消息
keep = [True] * len(self.messages)
current = total
for level in [MessageImportance.MINIMAL, MessageImportance.LOW]:
for i, msg in enumerate(self.messages):
if keep[i] and msg.importance == level:
keep[i] = False
current -= msg.token_estimate
if current <= budget:
break
if current <= budget:
break
return [self.messages[i].to_api_format()
for i in range(len(self.messages)) if keep[i]]
def get_stats(self) -> Dict[str, Any]:
total = sum(m.token_estimate for m in self.messages)
if self.system_prompt:
total += self.system_prompt.token_estimate
return {
"message_count": len(self.messages),
"total_estimated_tokens": total,
"window_capacity": self.max_tokens,
"utilization": f"{total / self.max_tokens * 100:.1f}%",
}
class VectorStore:
"""简单的向量存储"""
def __init__(self):
self.vectors: Dict[str, List[float]] = {}
self.entries: Dict[str, Any] = {}
def add(self, memory_id: str, vector: List[float], entry: Any):
self.vectors[memory_id] = vector
self.entries[memory_id] = entry
def search(self, query_vector: List[float], top_k: int = 5,
min_similarity: float = 0.2) -> List[tuple]:
import math
results = []
for memory_id, vector in self.vectors.items():
dot = sum(a * b for a, b in zip(query_vector, vector))
norm_a = math.sqrt(sum(a * a for a in query_vector))
norm_b = math.sqrt(sum(b * b for b in vector))
similarity = dot / (norm_a * norm_b) if norm_a > 0 and norm_b > 0 else 0
if similarity >= min_similarity:
results.append((self.entries[memory_id], similarity))
results.sort(key=lambda x: x[1], reverse=True)
return results[:top_k]
def size(self) -> int:
return len(self.entries)
class LongTermMemory:
"""长期记忆"""
def __init__(self, embedding_provider=None):
self.embedding = embedding_provider or self._default_embedding
self.store = VectorStore()
def _default_embedding(self, text: str) -> List[float]:
"""默认的简单嵌入（字符哈希）"""
vector = [0.0] * 64
for char in text.lower():
idx = ord(char) % 64
vector[idx] += 1.0
norm = sum(v * v for v in vector) ** 0.5
if norm > 0:
vector = [v / norm for v in vector]
return vector
def store(self, content: str, category: str = "general",
importance: int = 3, metadata: Dict[str, Any] = None) -> str:
import hashlib
memory_id = hashlib.md5(f"{category}:{content}".encode()).hexdigest()[:16]
if memory_id in self.store.entries:
existing = self.store.entries[memory_id]
existing["importance"] = min(5, existing.get("importance", 3) + 1)
existing["last_accessed_at"] = datetime.now().isoformat()
return memory_id
entry = {
"memory_id": memory_id,
"content": content,
"category": category,
"importance": importance,
"created_at": datetime.now().isoformat(),
"access_count": 0,
"metadata": metadata or {},
}
vector = self.embedding(content)
self.store.add(memory_id, vector, entry)
return memory_id
def retrieve(self, query: str, top_k: int = 5,
category: str = None) -> List[tuple]:
query_vector = self.embedding(query)
results = self.store.search(query_vector, top_k=top_k)
if category:
results = [(e, s) for e, s in results if e["category"] == category]
for entry, _ in results:
entry["access_count"] += 1
entry["last_accessed_at"] = datetime.now().isoformat()
return results
def retrieve_as_context(self, query: str, top_k: int = 3) -> str:
results = self.retrieve(query, top_k=top_k)
if not results:
return ""
lines = ["[相关的历史记忆]"]
for i, (entry, score) in enumerate(results, 1):
lines.append(f"{i}. {entry['content']}（相关度：{score:.2f}）")
lines.append("[历史记忆结束]")
return "\n".join(lines)
def get_stats(self) -> Dict[str, Any]:
total = self.store.size()
categories = {}
for entry in self.store.entries.values():
cat = entry["category"]
categories[cat] = categories.get(cat, 0) + 1
return {
"total_memories": total,
"categories": categories,
}
class SessionStateManager:
"""会话状态管理"""
def __init__(self):
self.sessions: Dict[str, Dict[str, Any]] = {}
self.storage_dir = "./session_data"
os.makedirs(self.storage_dir, exist_ok=True)
def create_session(self, user_id: str = None) -> str:
session_id = str(uuid.uuid4())[:12]
self.sessions[session_id] = {
"session_id": session_id,
"user_id": user_id,
"started_at": datetime.now().isoformat(),
"last_activity_at": datetime.now().isoformat(),
"conversation_turns": 0,
"confirmed_facts": {},
"current_subject": None,
"task_history": [],
}
return session_id
def get_session(self, session_id: str) -> Optional[Dict[str, Any]]:
return self.sessions.get(session_id)
def update(self, session_id: str, **kwargs):
if session_id in self.sessions:
self.sessions[session_id].update(kwargs)
self.sessions[session_id]["last_activity_at"] = datetime.now().isoformat()
self.sessions[session_id]["conversation_turns"] += 1
def save_to_disk(self, session_id: str):
if session_id in self.sessions:
filepath = os.path.join(self.storage_dir, f"{session_id}.json")
with open(filepath, 'w', encoding='utf-8') as f:
json.dump(self.sessions[session_id], f, ensure_ascii=False, indent=2)
def load_from_disk(self, session_id: str) -> Optional[Dict[str, Any]]:
filepath = os.path.join(self.storage_dir, f"{session_id}.json")
if os.path.exists(filepath):
with open(filepath, 'r', encoding='utf-8') as f:
data = json.load(f)
self.sessions[session_id] = data
return data
return None
def get_context_summary(self, session_id: str) -> str:
session = self.get_session(session_id)
if not session:
return ""
parts = []
if session.get("confirmed_facts"):
facts = [f"{k}={v}" for k, v in session["confirmed_facts"].items()]
parts.append(f"已确认信息：{'；'.join(facts)}")
if session.get("current_subject"):
subj = session["current_subject"]
parts.append(f"当前讨论对象：{subj.get('description', '')}")
if session.get("task_history"):
last_task = session["task_history"][-1]
parts.append(f"最近完成任务：{last_task.get('description', '')}")
return "\n".join(parts) if parts else ""第二步：构建完整的记忆系统
python
class MemorySystem:
"""完整的记忆系统"""
def __init__(self, system_prompt: str,
long_term_memory: Optional[LongTermMemory] = None):
self.system_prompt = system_prompt
self.long_term = long_term_memory or LongTermMemory()
self.state_manager = SessionStateManager()
self.session_id: Optional[str] = None
self.short_term: Optional[WorkingMemory] = None
self.last_user_input: str = ""
def start_session(self, user_id: str = None) -> str:
"""开始一个新会话"""
self.session_id = self.state_manager.create_session(user_id)
self.short_term = WorkingMemory(max_tokens=8000)
self.short_term.set_system_prompt(self.system_prompt)
return self.session_id
def resume_session(self, session_id: str) -> bool:
"""恢复之前的会话"""
session = self.state_manager.load_from_disk(session_id)
if not session:
return False
self.session_id = session_id
self.short_term = WorkingMemory(max_tokens=8000)
self.short_term.set_system_prompt(self.system_prompt)
# 注入会话状态摘要
summary = self.state_manager.get_context_summary(session_id)
if summary:
self.short_term.add_message(
"user", f"[恢复会话]\n{summary}",
importance=MessageImportance.HIGH
)
return True
def add_user_message(self, content: str):
"""添加用户消息，并检索长期记忆"""
self.last_user_input = content
# 添加用户消息
self.short_term.add_message("user", content,
importance=MessageImportance.HIGH)
# 检索长期记忆
relevant = self.long_term.retrieve(content, top_k=3)
if relevant:
context = self.long_term.retrieve_as_context(content, top_k=3)
self.short_term.add_message(
"system", context,
importance=MessageImportance.HIGH
)
# 从用户消息中提取并存储记忆
self._extract_and_store_memory(content, source="user_input")
def add_assistant_message(self, content: str):
"""添加助手消息"""
self.short_term.add_message("assistant", content,
importance=MessageImportance.MEDIUM)
def add_tool_result(self, tool_name: str, content: str,
tool_call_id: str = ""):
"""添加工具结果，并提取记忆"""
self.short_term.add_message(
"tool", content,
importance=MessageImportance.HIGH,
tool_call_id=tool_call_id,
)
# 从工具结果中提取记忆
self._extract_and_store_memory(
f"[{tool_name}] {content[:200]}",
source="tool_result"
)
def _extract_and_store_memory(self, content: str, source: str):
"""提取并存储长期记忆"""
# 偏好类
if "喜欢" in content or "偏好" in content or "习惯" in content:
self.long_term.store(
content=f"用户偏好：{content[:100]}",
category="user_preference",
importance=4,
metadata={"source": source},
)
# 事实类
if "我是" in content or "我在" in content or "我的" in content:
self.long_term.store(
content=f"用户信息：{content[:100]}",
category="fact",
importance=4,
metadata={"source": source},
)
def get_context(self) -> List[Dict[str, Any]]:
"""获取当前上下文"""
return self.short_term.get_context() if self.short_term else []
def end_session(self):
"""结束会话，保存状态"""
if self.session_id:
self.state_manager.save_to_disk(self.session_id)
def get_stats(self) -> Dict[str, Any]:
stats = {}
if self.short_term:
stats["short_term"] = self.short_term.get_stats()
stats["long_term"] = self.long_term.get_stats()
return stats第三步：构建带持久记忆的Agent
python
class PersistentAgent:
"""带持久记忆的Agent"""
def __init__(self, api_key: str, model: str = "gpt-4o-mini"):
from openai import OpenAI
self.client = OpenAI(api_key=api_key, base_url=os.getenv("LLM_BASE_URL", "https://api.openai.com/v1"))
self.model = model
self.memory = MemorySystem(
system_prompt=(
"你是一个个人助理Agent。你可以帮助用户：\n"
"1. 查询日历和日程\n"
"2. 创建和管理任务\n"
"3. 记录用户偏好和习惯\n"
"4. 执行简单计算\n\n"
"行为准则：\n"
"- 记住用户的偏好，在后续对话中使用\n"
"- 如果信息不足，先询问\n"
"- 回答简洁，不超过100字\n"
)
)
self.tools = self._create_tools()
def _create_tools(self) -> Dict[str, Any]:
"""创建工具集"""
return {
"query_calendar": {
"description": "查询指定日期的日程安排。参数date格式YYYY-MM-DD。",
"func": self._query_calendar,
},
"create_task": {
"description": "创建新任务。参数title是任务标题，due_date是截止日期。",
"func": self._create_task,
},
"calculate": {
"description": "计算数学表达式。参数expression是表达式。",
"func": self._calculate,
},
}
def _query_calendar(self, date: str) -> str:
events = {
"2026-09-07": "10:00 产品评审会",
"2026-09-08": "14:00 周会",
"2026-09-09": "09:30 一对一沟通",
}
return events.get(date, f"{date} 没有日程安排。")
def _create_task(self, title: str, due_date: str) -> str:
return f"任务已创建：{title}（截止：{due_date}）"
def _calculate(self, expression: str) -> str:
try:
result = eval(expression, {"__builtins__": {}}, {})
return f"计算结果：{result}"
except Exception as e:
return f"计算出错：{e}"
def chat(self, user_input: str, session_id: str = None) -> str:
"""处理用户输入，返回Agent响应"""
# 管理会话
if session_id:
if not self.memory.resume_session(session_id):
self.memory.start_session()
elif not self.memory.session_id:
self.memory.start_session()
# 添加用户消息
self.memory.add_user_message(user_input)
# 构建工具Schema
tool_schemas = []
for name, tool in self.tools.items():
tool_schemas.append({
"type": "function",
"function": {
"name": name,
"description": tool["description"],
"parameters": {
"type": "object",
"properties": self._infer_parameters(tool["func"]),
"required": self._infer_required(tool["func"]),
}
}
})
# Agent循环
max_iterations = 10
for iteration in range(max_iterations):
response = self.client.chat.completions.create(
model=self.model,
messages=self.memory.get_context(),
tools=tool_schemas,
temperature=0.0,
)
message = response.choices[0].message
if message.tool_calls:
self.memory.add_assistant_message(
f"[调用工具] {message.tool_calls[0].function.name}"
)
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
if name in self.tools:
result = self.tools[name]["func"](**args)
self.memory.add_tool_result(name, result, tool_call.id)
else:
answer = message.content or ""
self.memory.add_assistant_message(answer)
return answer
return "已达到最大循环次数，未能完成任务。"
def _infer_parameters(self, func) -> Dict[str, Any]:
import inspect
sig = inspect.signature(func)
properties = {}
for param_name, param in sig.parameters.items():
properties[param_name] = {
"type": "string",
"description": f"参数 {param_name}",
}
return properties
def _infer_required(self, func) -> List[str]:
import inspect
sig = inspect.signature(func)
return [p for p in sig.parameters if sig.parameters[p].default == inspect.Parameter.empty]
def get_memory_stats(self) -> Dict[str, Any]:
"""获取记忆统计"""
return self.memory.get_stats()
def save_session(self):
"""保存当前会话"""
self.memory.end_session()
return self.memory.session_id第四步：测试Agent的持久记忆
python
def test_persistent_agent():
"""测试带持久记忆的Agent"""
agent = PersistentAgent(api_key=os.getenv("OPENAI_API_KEY", ""))
print("=" * 50)
print("测试1：偏好记忆")
print("=" * 50)
# 第一轮对话：告诉Agent偏好
response = agent.chat("我比较喜欢在下午开会，上午通常比较忙。")
print(f"用户：我比较喜欢在下午开会，上午通常比较忙。")
print(f"Agent：{response}\n")
# 第二轮对话：Agent应该记住偏好
response = agent.chat("帮我安排一个会议")
print(f"用户：帮我安排一个会议")
print(f"Agent：{response}\n")
# 保存会话
session_id = agent.save_session()
print(f"会话已保存：{session_id}\n")
print("=" * 50)
print("测试2：跨会话记忆")
print("=" * 50)
# 模拟新的Agent实例（如服务重启）
agent2 = PersistentAgent(api_key=os.getenv("OPENAI_API_KEY", ""))
# 新会话中，Agent应该仍然记得偏好
response = agent2.chat("帮我安排一个和产品组的会议", session_id=session_id)
print(f"用户（新会话）：帮我安排一个和产品组的会议")
print(f"Agent：{response}\n")
print("=" * 50)
print("记忆统计")
print("=" * 50)
stats = agent2.get_memory_stats()
print(json.dumps(stats, ensure_ascii=False, indent=2))预期输出
==================================================
测试1：偏好记忆
==================================================
用户：我比较喜欢在下午开会，上午通常比较忙。
Agent：好的，我记住了。您喜欢在下午开会，上午通常比较忙。
用户：帮我安排一个会议
Agent：好的，考虑到您偏好下午开会，我建议安排在下午。请问会议的主题是什么，您希望在哪天举行？
会话已保存：a1b2c3d4e5f6
==================================================
测试2：跨会话记忆
==================================================
用户（新会话）：帮我安排一个和产品组的会议
Agent：好的，我记得您偏好下午开会。让我为您安排一个与产品组的会议。请问您希望在哪天举行？
==================================================
记忆统计
==================================================
{
"short_term": {
"message_count": 6,
"total_estimated_tokens": 450,
"window_capacity": 8000,
"utilization": "5.6%"
},
"long_term": {
"total_memories": 3,
"categories": {
"user_preference": 1,
"fact": 2
}
}
}观察：Agent如何利用持久记忆
在测试中，Agent展现了几个关键能力：
第一：偏好记忆。 用户说"我比较喜欢在下午开会"后，Agent将其存储为长期记忆。在后续对话中，Agent主动提出了"考虑您偏好下午开会"的建议。
第二：跨会话记忆。 即使创建了新的Agent实例（模拟服务重启），通过恢复会话，Agent仍然记得用户的偏好。这是长期记忆和会话持久化的共同作用。
第三：记忆的渐进积累。 随着用户与Agent的互动增多，长期记忆中的条目不断增加，Agent对用户的理解也越来越深入。这就是"越用越懂你"的效果。
一个值得注意的设计细节
在MemorySystem.add_user_message中，长期记忆的检索结果被注入为system消息而非user消息。这很重要：
system消息的语义是"给LLM的额外指示/背景信息"
user消息的语义是"用户说的话"
历史记忆是背景信息，不是用户当前说的话。将它们注入为system消息，能让LLM正确理解：这些是"参考信息"，而非"用户的新指令"。
本节小结
本节将前面各节构建的组件整合为一个完整的、带持久记忆的Agent：
短期记忆：管理当前会话的上下文窗口
长期记忆：跨会话存储用户偏好和事实
会话状态：持久化会话信息，支持恢复
整合效果：Agent在多次交互中变得越来越"懂"用户
核心价值：持久记忆让Agent从"一次性工具"变成了"持续陪伴的助手"。用户不需要每次重复告诉Agent自己的偏好和习惯——Agent记得。
第5章到此结束。在下一章中，我们将讨论Agent推理与规划——如何让Agent不仅记住信息，还能有效地分解任务、制定计划、在执行中反思和调整。