《Agent Harness入门教程》

---

1 AI Agent与Harness概述

---

2 核心概念与架构模式

---

3 环境搭建与基础组件

---

4 工具系统设计

---

5 记忆与会话管理

---

6 Agent推理与规划

---

7 商业项目实战练习

---

---

短期记忆的边界

在5.1节中，你构建了短期记忆系统——它管理着当前会话的上下文窗口。但短期记忆有一个根本性的局限：会话结束后，一切都被清空。

用户昨天告诉Agent"我喜欢在下午开会"，今天Agent完全不记得。用户上周让Agent查过某个项目的文档，这周再问相关问题时，Agent需要从头查起。

这不是因为Agent"不够聪明"，而是因为短期记忆的物理边界就是当前会话的上下文窗口。要突破这个边界，你需要一个不同的记忆机制——长期记忆。

长期记忆解决的核心问题是：如何让Agent跨会话记住信息，并在需要时高效地检索出来。

长期记忆与短期记忆的本质区别

维度	短期记忆	长期记忆
生命周期	当前会话	跨会话持久
存储位置	上下文窗口	外部存储（向量库、数据库）
访问方式	直接可见	需要主动检索
容量	有限（窗口大小）	理论上无限
速度	即时	有检索延迟
精度	完整保留	可能检索到不相关内容

关键的架构差异：短期记忆是推送给LLM的（LLM被动看到所有内容），长期记忆是按需检索的（LLM主动查询相关内容）。这意味着长期记忆系统需要解决一个额外的问题：如何知道什么信息与当前任务相关？

向量数据库的基本原理

向量数据库是长期记忆最常用的存储后端。它的核心思想是：将文本转换为向量（一组数字），然后用向量相似度来衡量"语义相关性"。

文本 → 嵌入模型 → 向量 → 存储
                         ↓
查询 → 嵌入模型 → 查询向量 → 相似度检索 → 相关文本

嵌入（Embedding）是将文本映射到高维向量空间的过程。语义相近的文本，其向量在空间中距离也相近。

例如：

• "用户喜欢在下午开会" → [0.12, -0.45, 0.78, ...]
• "用户的会议时间偏好是下午" → [0.11, -0.43, 0.79, ...]（向量相近）

这两句话虽然措辞不同，但语义相近，它们的向量在空间中距离很近。向量检索就是利用这个特性，找到与查询"语义相关"而不仅仅是"字面匹配"的内容。

构建一个长期记忆系统

第一步：定义记忆条目的数据结构

from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional
from datetime import datetime
import json
import hashlib

@dataclass
class MemoryEntry:
    """一条长期记忆"""
    memory_id: str
    content: str                    # 记忆的文本内容
    category: str = "general"       # 分类：user_preference, fact, event, knowledge
    source: str = "conversation"    # 来源：哪次对话、哪个工具
    created_at: str = field(default_factory=lambda: datetime.now().isoformat())
    last_accessed_at: Optional[str] = None
    access_count: int = 0
    importance: int = 3             # 1-5，重要性评分
    metadata: Dict[str, Any] = field(default_factory=dict)

    def to_text(self) -> str:
        """转换为可嵌入的文本"""
        return self.content

    def to_dict(self) -> Dict[str, Any]:
        return {
            "memory_id": self.memory_id,
            "content": self.content,
            "category": self.category,
            "source": self.source,
            "created_at": self.created_at,
            "importance": self.importance,
            "metadata": self.metadata,
        }


def generate_memory_id(content: str, category: str) -> str:
    """生成记忆ID（基于内容的哈希，用于去重）"""
    hash_input = f"{category}:{content}"
    return hashlib.md5(hash_input.encode('utf-8')).hexdigest()[:16]

第二步：实现嵌入层

嵌入是长期记忆系统的基础。你可以使用商业API（如OpenAI的Embeddings）或本地模型：

class EmbeddingProvider:
    """嵌入提供者：将文本转换为向量"""

    def __init__(self, client=None, model: str = "text-embedding-3-small"):
        self.client = client
        self.model = model

    def embed(self, text: str) -> List[float]:
        """将文本转换为向量"""
        if self.client is None:
            return self._fallback_embed(text)

        try:
            response = self.client.embeddings.create(
                model=self.model,
                input=text[:8000],  # 限制输入长度
            )
            return response.data[0].embedding
        except Exception as e:
            print(f"嵌入失败，使用降级方案：{e}")
            return self._fallback_embed(text)

    def _fallback_embed(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

第三步：实现向量存储与检索

不依赖外部向量数据库，用纯Python实现一个简单的向量存储：

import math
from typing import List, Tuple

class VectorStore:
    """简单的向量存储与检索"""

    def __init__(self):
        self.vectors: Dict[str, List[float]] = {}
        self.entries: Dict[str, MemoryEntry] = {}

    def add(self, memory_id: str, vector: List[float], entry: MemoryEntry):
        """添加一个向量和对应的记忆条目"""
        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.3) -> List[Tuple[MemoryEntry, float]]:
        """检索与查询向量最相似的记忆"""
        results = []

        for memory_id, vector in self.vectors.items():
            similarity = self._cosine_similarity(query_vector, vector)
            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 _cosine_similarity(self, vec_a: List[float], vec_b: List[float]) -> float:
        """计算余弦相似度"""
        dot_product = sum(a * b for a, b in zip(vec_a, vec_b))
        norm_a = math.sqrt(sum(a * a for a in vec_a))
        norm_b = math.sqrt(sum(b * b for b in vec_b))

        if norm_a == 0 or norm_b == 0:
            return 0.0

        return dot_product / (norm_a * norm_b)

    def remove(self, memory_id: str):
        """删除一条记忆"""
        self.vectors.pop(memory_id, None)
        self.entries.pop(memory_id, None)

    def size(self) -> int:
        return len(self.entries)

第四步：组装完整的长期记忆系统

class LongTermMemory:
    """长期记忆系统"""

    def __init__(self, embedding_provider: EmbeddingProvider, 
                 vector_store: VectorStore = None):
        self.embedding = embedding_provider
        self.store = vector_store or VectorStore()

    def store(self, content: str, category: str = "general",
              importance: int = 3, metadata: Dict[str, Any] = None) -> str:
        """存储一条记忆"""
        # 生成记忆ID（用于去重）
        memory_id = generate_memory_id(content, category)

        # 检查是否已存在
        if memory_id in self.store.entries:
            # 已存在：更新重要性（累加）
            existing = self.store.entries[memory_id]
            existing.importance = min(5, existing.importance + 1)
            existing.last_accessed_at = datetime.now().isoformat()
            return memory_id

        # 创建新记忆
        entry = MemoryEntry(
            memory_id=memory_id,
            content=content,
            category=category,
            importance=importance,
            metadata=metadata or {},
        )

        # 嵌入并存储
        vector = self.embedding.embed(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[MemoryEntry, float]]:
        """检索与查询相关的记忆"""
        query_vector = self.embedding.embed(query)
        results = self.store.search(query_vector, top_k=top_k)

        # 按分类过滤
        if category:
            results = [(entry, score) for entry, score in results 
                       if entry.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 = 5) -> str:
        """检索并将结果格式化为上下文字符串"""
        results = self.retrieve(query, top_k=top_k)

        if not results:
            return ""

        lines = ["[以下是相关的历史记忆]"]
        for i, (entry, similarity) in enumerate(results, 1):
            lines.append(f"{i}. {entry.content}（相关度：{similarity:.2f}，分类：{entry.category}）")
        lines.append("[历史记忆结束]")

        return "\n".join(lines)

    def get_stats(self) -> Dict[str, Any]:
        """获取记忆系统的统计信息"""
        total = self.store.size()
        categories = {}
        total_access = 0

        for entry in self.store.entries.values():
            categories[entry.category] = categories.get(entry.category, 0) + 1
            total_access += entry.access_count

        return {
            "total_memories": total,
            "categories": categories,
            "total_access_count": total_access,
            "avg_access_per_memory": total_access / total if total > 0 else 0,
        }

何时写入长期记忆

一个关键的设计问题是：什么信息值得被存入长期记忆？ 不是所有对话内容都值得长期保存。

以下信息值得存储：

class MemoryExtractor:
    """从对话中提取值得长期记忆的内容"""

    def extract_from_user_message(self, content: str) -> List[Dict[str, Any]]:
        """从用户消息中提取候选记忆"""
        candidates = []

        # 偏好类信息
        preference_patterns = [
            r"我(?:喜欢|偏好|习惯|通常|一般)(?:在|用|选择|是)?(.{2,50})",
            r"不要(.{2,50})",
            r"我(?:总是|从不|经常)(.{2,50})",
        ]

        import re
        for pattern in preference_patterns:
            matches = re.findall(pattern, content)
            for match in matches:
                candidates.append({
                    "content": f"用户偏好：{match.strip()}",
                    "category": "user_preference",
                    "importance": 4,
                })

        # 事实类信息
        fact_patterns = [
            r"我(?:是|在|住|工作)(?:在|于)?(.{2,50})",
            r"我(?:的电话|邮箱|地址)(?:是|为)?(.{2,50})",
        ]

        for pattern in fact_patterns:
            matches = re.findall(pattern, content)
            for match in matches:
                candidates.append({
                    "content": f"用户信息：{match.strip()}",
                    "category": "fact",
                    "importance": 4,
                })

        return candidates

    def extract_from_tool_result(self, tool_name: str, result: str) -> List[Dict[str, Any]]:
        """从工具结果中提取候选记忆"""
        candidates = []

        # 工具结果的摘要信息
        if len(result) > 100:
            candidates.append({
                "content": f"[{tool_name}] {result[:200]}",
                "category": "knowledge",
                "importance": 2,
                "metadata": {"tool": tool_name},
            })

        return candidates

    def should_store(self, content: str, category: str, 
                     existing_memories: List[MemoryEntry]) -> bool:
        """判断是否应该存储这条记忆"""
        # 太短的记忆不值得存储
        if len(content) < 10:
            return False

        # 与已有记忆高度重复的不存储
        for existing in existing_memories:
            if existing.category == category:
                # 简单的相似度判断
                overlap = self._text_overlap(content, existing.content)
                if overlap > 0.7:
                    return False

        return True

    def _text_overlap(self, text_a: str, text_b: str) -> float:
        """计算两段文本的重叠度"""
        words_a = set(text_a)
        words_b = set(text_b)
        if not words_a or not words_b:
            return 0.0
        intersection = len(words_a & words_b)
        union = len(words_a | words_b)
        return intersection / union if union > 0 else 0.0

将长期记忆集成到Agent循环中

更新MemorySystem，让它同时管理短期和长期记忆：

class MemorySystem:
    """完整的记忆系统：短期 + 长期"""

    def __init__(self, system_prompt: str, 
                 long_term_memory: Optional[LongTermMemory] = None,
                 embedding_provider: Optional[EmbeddingProvider] = None):
        self.system_prompt = system_prompt
        self.short_term = WorkingMemory(max_tokens=8000)
        self.short_term.set_system_prompt(system_prompt)

        self.long_term = long_term_memory or (
            LongTermMemory(embedding_provider) if embedding_provider else None
        )

        self.extractor = MemoryExtractor()
        self.messages: List[Dict[str, Any]] = []

    def initialize(self, user_input: str):
        """初始化一次新的会话"""
        # 清空短期记忆
        self.short_term = WorkingMemory(max_tokens=8000)
        self.short_term.set_system_prompt(self.system_prompt)

        # 添加用户输入
        self.short_term.add_user_message(user_input)

        # 检查长期记忆中是否有相关信息
        if self.long_term:
            relevant = self.long_term.retrieve(user_input, top_k=3)
            if relevant:
                context_text = self.long_term.retrieve_as_context(user_input, top_k=3)
                # 将相关记忆注入上下文
                self.short_term.add_user_message(
                    f"[系统注入了相关的历史记忆]\n{context_text}\n"
                    f"请在有帮助的情况下使用这些历史信息。"
                )

        self.messages = self.short_term.get_context()

        # 从用户输入中提取并存储长期记忆
        if self.long_term:
            candidates = self.extractor.extract_from_user_message(user_input)
            for candidate in candidates:
                if self.extractor.should_store(
                    candidate["content"], 
                    candidate["category"],
                    [entry for entry, _ in self.long_term.retrieve(candidate["content"], top_k=3)]
                ):
                    self.long_term.store(
                        content=candidate["content"],
                        category=candidate["category"],
                        importance=candidate["importance"],
                    )

    def add_tool_result(self, result):
        """添加工具结果到短期记忆，并提取长期记忆"""
        self.short_term.add_tool_result(
            result.tool_call_id if hasattr(result, 'tool_call_id') else "",
            result.content
        )

        # 从工具结果中提取长期记忆
        if self.long_term:
            candidates = self.extractor.extract_from_tool_result(
                getattr(result, 'tool_name', 'unknown'),
                result.content
            )
            for candidate in candidates:
                if self.extractor.should_store(
                    candidate["content"],
                    candidate["category"],
                    [entry for entry, _ in self.long_term.retrieve(candidate["content"], top_k=3)]
                ):
                    self.long_term.store(
                        content=candidate["content"],
                        category=candidate["category"],
                        importance=candidate["importance"],
                        metadata=candidate.get("metadata", {}),
                    )

        self.messages = self.short_term.get_context()

    def get_context(self) -> List[Dict[str, Any]]:
        """获取当前上下文"""
        return self.short_term.get_context()

    def get_stats(self) -> Dict[str, Any]:
        """获取完整的记忆统计"""
        stats = {
            "short_term": self.short_term.get_stats(),
        }
        if self.long_term:
            stats["long_term"] = self.long_term.get_stats()
        return stats

一个完整的示例

# examples/long_term_memory_demo.py

from openai import OpenAI

# 创建嵌入提供者
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY", ""))
embedding_provider = EmbeddingProvider(client=client)

# 创建长期记忆
ltm = LongTermMemory(embedding_provider)

# 存储一些记忆
ltm.store("用户偏好：喜欢在下午开会", category="user_preference", importance=5)
ltm.store("用户信息：产品经理，负责搜索功能", category="fact", importance=4)
ltm.store("项目：Q3的重点是提升搜索相关性和用户体验", category="knowledge", importance=3)

# 检索测试
print("检索'会议时间'：")
results = ltm.retrieve("什么时候开会比较好", top_k=3)
for entry, score in results:
    print(f"  [{score:.2f}] {entry.content}")

print("\n检索'用户的工作'：")
results = ltm.retrieve("用户是做什么的", top_k=3)
for entry, score in results:
    print(f"  [{score:.2f}] {entry.content}")

# 集成到MemorySystem
memory_system = MemorySystem(
    system_prompt="你是一个智能助手。",
    long_term_memory=ltm,
)

# 模拟一次新会话
memory_system.initialize("帮我安排一个会议")
context = memory_system.get_context()
print(f"\n会话上下文包含{len(context)}条消息")
for msg in context:
    print(f"  [{msg['role']}] {msg['content'][:80]}...")

长期记忆的挑战与权衡

挑战一：检索的相关性

向量检索可能返回语义相关但实际无用的信息。用户问"帮我查一下订单"，检索可能返回"用户上次问过订单状态"的记忆——这条记忆对当前任务没有帮助，但占据了宝贵的上下文空间。

应对：设置相似度阈值，只注入高于阈值的记忆。在实践中，阈值需要根据场景调优。

挑战二：记忆的时效性

用户说"我喜欢用Python"，但一年后可能已经转向了其他语言。长期记忆中过时的信息可能误导Agent的判断。

应对：在记忆条目中记录last_accessed_at和access_count，对长期未被访问的记忆降低重要性。在检索时，可以给新近的记忆更高的权重。

挑战三：记忆的膨胀

随着时间推移，长期记忆会不断增长。如果不去管理，存储量会越来越大，检索也越来越慢。

应对：

• 定期清理：删除长期未被访问且重要性低的记忆
• 记忆合并：将相似的记忆合并为一条
• 分类归档：将不常用的记忆移到"冷存储"

挑战四：存储与检索的成本

每次嵌入和检索都有计算成本（API调用或本地计算）。高频的检索会显著增加Agent的执行成本。

应对：

• 只在会话开始时检索一次，而非每轮循环都检索
• 缓存查询向量，避免重复嵌入
• 设置检索触发条件（如仅当用户消息包含某些关键词时）

本节小结

长期记忆让Agent突破会话的边界，跨时间记住信息。本节构建了一个基于向量检索的长期记忆系统：

• 存储：将文本嵌入为向量，存入向量存储
• 检索：将查询嵌入为向量，通过余弦相似度找到相关记忆
• 集成：在会话开始时注入相关记忆，在对话过程中提取并存储新记忆
• 管理：去重、重要性评分、访问统计

核心原则：长期记忆不是"存得越多越好"，而是"在需要时能检索到最相关的信息"。存储什么、何时检索、检索多少——这些决策比存储技术本身更影响Agent的表现。

在下一节中，我们将讨论会话状态管理——如何管理多轮对话的连续性和一致性。

---

---