企业官网建设流程全解析

前段时间我在做一个带长期对话能力的 Agent 项目，碰到一个特别典型的问题：模型每次都是“冷启动”，用户上一轮刚说过的偏好，这一轮又要重新问一遍；用户昨天交代过的关键信息，今天再聊时模型完全没反应。说白了，大模型本身是不带记忆的，所有的智能都依赖于当下这个请求里塞了多少上下文。而 Agent 要真正像一个“助手”而不是“接口”，Memory 模块是绕不开的一块硬骨头。

这篇文章就围绕我给 Agent 装记忆的完整过程来写。我会从最简单的 InMemory 内存实现开始，先让流程跑通，再落到文件持久化，解决“进程重启就失忆”的问题，最后升级到 Milvus 向量数据库，实现跨会话的语义召回。整个过程我会把每一步的设计思路、代码实现、踩过的坑都摊开讲，适合正在做 Agent 开发、想在项目里接入记忆能力的同学参考。不管你之前有没有接触过向量数据库，照着做就能跑起来。

1. Agent Memory 到底在解决什么问题

1.1 大模型天生没有记忆

先说本质问题：大模型 API 是一个无状态函数。你传一段输入，它给你一段输出，仅此而已。它不会记得五分钟前你问过什么，更不会记得昨天你让它记过什么。这就像一个每天上班都换一个新同事的团队，你每次对接的人都对你的项目一无所知，你必须把背景从头讲一遍。

有人会说，那我直接把历史对话拼进 prompt 不就行了？确实可以，但有两个硬限制。第一个是上下文窗口有限，几百 K 的上下文也不可能无限塞；第二个是无效信息会稀释注意力，你塞进去 10 万 token 的历史记录，模型找关键信息的难度反而会变大，回答质量会下降，响应延迟和成本也会一起涨。

所以 Memory 模块的核心职责，不是“存储”本身，而是“在合适的时候把合适的记忆放回上下文里”。存储只是手段，检索和组装上下文才是目的。想清楚这一点，后面每一步设计都不会走偏。

1.2 Memory 的三种类型与分工

实际工程里，Memory 不是单一组件，而是分层的。我习惯把它拆成三种：

这三个层次不是互相替代的关系，而是配合使用。工作记忆解决的是“眼下聊到哪了”，长期记忆解决的是“这个人以前说过什么”，语义记忆解决的是“和当前问题最相关的背景知识是什么”。我后面要讲的从 InMemory 到文件到 Milvus 的演进，本质上就是一层一层补齐这几种记忆能力。

1.3 Memory 模块在 Agent 中的位置

在 Agent 的典型架构里，Memory 模块通常不是独立服务，而是编排层的一部分。LLM 是大脑，工具是手脚，Memory 则是大脑的“海马体”。

具体来说，在每次调用 LLM 之前，Agent 会先走一遍“记忆检索”流程：从短期缓存里取最近的对话，从长期存储里按当前 query 召回相关记忆，然后把这两部分拼进 system prompt 或者作为额外的 context 注入。LLM 生成回复之后，又会把这一段交互写入记忆系统。这个“先读后写”的流程，就是 Memory 模块存在的意义。

2. 第一层：InMemory 内存实现，先让流程跑起来

2.1 一个最简 Memory 类的写法

很多 Agent 项目一开始根本不考虑持久化，先用一个列表把历史消息存下来。这个阶段最重要的不是“存得久”，而是“接口要清晰”。我见过不少项目在早期随手写一堆全局变量来存对话，结果后面重构的时候到处找引用，非常痛苦。

我的建议是，哪怕只是内存存储，也把它封装成一个独立类，定义好 add、get_recent、build_context 这几个方法。这样后面替换成文件存储或者向量库存储时，上层代码几乎不用动。一个最小实现长这样：

from typing import List, Dict from dataclasses import dataclass, field import time @dataclass class Message: role: str content: str timestamp: float = field(default_factory=time.time) class InMemoryMemory: def __init__(self, max_messages: int = 20): self.messages: List[Message] = [] self.max_messages = max_messages def add(self, role: str, content: str) -> None: self.messages.append(Message(role=role, content=content)) if len(self.messages) > self.max_messages: self.messages = self.messages[-self.max_messages:] def get_recent(self, n: int = 10) -> List[Message]: return self.messages[-n:] def build_context(self, n: int = 10) -> str: recent = self.get_recent(n) return "\n".join(f"{msg.role}: {msg.content}" for msg in recent)

这里有两个设计细节值得说。第一，我用timestamp字段给每条消息打点，这个字段在内存阶段看似没用，但后面做文件持久化、按时间过滤时都用得上。尽早把时间维度纳入数据结构，后期会省很多事。第二，max_messages用滑动窗口裁剪，超出窗口就把最早的消息丢掉。这里要注意，裁剪时保留的是末尾的最近 N 条，因为新消息永远追加在列表末尾。

2.2 什么时候 InMemory 是够用的

不是所有场景都需要上向量数据库。如果你的 Agent 只做单轮工具调用、或者对话轮次很少、或者用户压根就没有跨会话恢复上下文的需求，那 InMemory 完全够用。

我记得早期做一个“表格问答 Agent”时，用户只会针对一张表连续问三到五个问题，问完就结束，不存在“下次再来”的场景。这个阶段用 InMemory 不仅简单，而且响应快，连序列化开销都省了。

另外，很多 Agent 框架本身也提供了内置的会话记忆实现。比如 LangChain 的ConversationBufferMemory就是典型的内存窗口实现，它的作用范围也仅限于当前进程内的会话管理。如果你的项目能接受“重启即失忆、多实例不共享”这两个前提，那直接用框架自带的内存实现就够了，没必要自己造轮子。

2.3 InMemory 的硬伤

InMemory 最大的问题有三个。

第一个是进程重启后记忆全丢。开发环境里无所谓，但生产环境只要部署实例一滚动更新，所有会话状态就归零了，用户体验直接退回解放前。

第二个是多实例部署时数据不共享。你有三个 Agent 实例在跑负载均衡，用户第一次请求打到实例 A，第二次请求打到实例 B，B 手里没有 A 存的记忆，于是又变成“冷启动”。这个问题不解决，InMemory 永远只能活在单机演示环境里。

第三个是上下文膨胀。滑动窗口只限制条数，不限制 token。如果某条消息内容特别长，20 条消息可能就能顶爆一次请求的上下文。后面真正做长期记忆时，一定要考虑按 token 而不是按条数来裁剪。

3. 第二层：文件落地，把记忆沉淀下来

3.1 用 JSONL 做追加式持久化

既然内存靠不住，那就落盘。最简单的持久化方式是把对话按行追加到 JSONL 文件里。JSONL 比 JSON 好在它是 append-only 的，每写一行都是一次独立操作，即使中途崩溃，最多丢最后一行，不会整个文件损坏。

我在项目里一般按 conversation_id 分文件存储，一个会话一个文件。实现起来非常直接：

import json import time from pathlib import Path from typing import List, Dict class FileMemory: def __init__(self, storage_dir: str = "./memory_store"): self.storage_dir = Path(storage_dir) self.storage_dir.mkdir(parents=True, exist_ok=True) def _path_for(self, conversation_id: str) -> Path: safe_name = "".join(c for c in conversation_id if c.isalnum() or c in "-_") return self.storage_dir / f"{safe_name}.jsonl" def append(self, conversation_id: str, role: str, content: str) -> None: record = { "role": role, "content": content, "timestamp": time.time(), } path = self._path_for(conversation_id) with open(path, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") def load(self, conversation_id: str, limit: int = 20) -> List[Dict]: path = self._path_for(conversation_id) if not path.exists(): return [] records = [] with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if line: records.append(json.loads(line)) return records[-limit:]

这里有两个地方是实践后才知道要处理的。第一是文件名转义。conversation_id 如果由用户传入，可能带/或..，直接拼到路径里会有安全风险，所以要做一次字符白名单过滤。第二是读取时对空行做防御，文件尾部可能因为上次写入中断而残留半截行，json.loads会直接抛异常。

3.2 用 SQLite 替代散文件

JSONL 文件用了一段时间，你会发现两个痛点。一是文件数量越来越多，目录管理变得麻烦；二是按条件检索非常弱，比如要查“某个用户今天的所有对话”，你得遍历所有文件再做字符串匹配。

这个阶段我推荐换成 SQLite。它本质还是单文件存储，不需要额外部署服务，但提供了 SQL 能力，事务、索引、条件查询都齐了。一个简单的表结构就够了：

CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, conversation_id TEXT NOT NULL, user_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, created_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_conversation ON messages(conversation_id, created_at); CREATE INDEX IF NOT EXISTS idx_user ON messages(user_id, created_at);

用 SQLite 之后，按会话查、按用户查、按时间范围查都变成了普通的 SELECT 语句，逻辑清晰很多。Python 里直接用内置的sqlite3模块就行，不需要引入 ORM。

3.3 文件存储解决不了“语义召回”

文件存储解决了“持久化”问题，但没有解决“召回”问题。它只是在记录层面做到了不丢数据，但检索方式仍然停留在“按会话 id 精确匹配”或者“关键词 like 匹配”这个级别。

什么叫语义召回？举个例子。用户周一在对话里说过“我平时通勤单程大概四十分钟”，周五他问“帮我估算下每周通勤时间”。用关键词检索根本匹配不上，因为“通勤四十分钟”和“每周通勤时间”之间没有一个共同的关键词，但语义上它们强相关。

这就是必须要引入向量数据库的原因。文件存储负责“把记忆记下来”，向量检索负责“把相关记忆找回来”。两者不是替代关系，而是互补关系。我后来的方案就是：SQLite 保留原始记录，Milvus 存向量索引，两边用同一条主键关联。

4. 第三层：上 Milvus 向量数据库，让记忆可召回

4.1 为什么要向量化存储

向量化存储的核心思路是：把一段文本转换成一组高维浮点数，用余弦相似度或内积来衡量文本之间的语义距离。相似的内容在向量空间里离得近，不相关的内容离得远。这样我们就能做到“我不管字面上是否包含同一个词，只要语义相关，就能召回”。

这个阶段要先解决 embedding 模型选型。我在实际项目里试过几种：OpenAI 的text-embedding-3-small效果好、维度适中，适合快速验证；但如果要本地部署、不想引入外部依赖，可以用 BGE-M3 或者bge-large-zh-v1.5，对中文支持也不错。维度决定了后面 Milvus 里 collection 的向量字段长度，所以先定 embedding 模型，再定 collection schema。

4.2 Milvus 单机版安装与启动

Milvus 是一个云原生向量数据库，生产环境一般用集群模式，但本地开发和中小项目直接用 Docker Compose 启动单机版就够了。

官方推荐的单机版部署包含三个组件：Milvus 主服务、etcd（元数据存储）、MinIO（对象存储）。用 Docker Compose 一次性启动：

version: '3.5' services: etcd: container_name: milvus-etcd image: quay.io/coreos/etcd:v3.5.5 environment: - ETCD_AUTO_COMPACTION_MODE=revision - ETCD_AUTO_COMPACTION_RETENTION=1000 - ETCD_QUOTA_BACKEND_BYTES=4294967296 volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/etcd:/etcd command: etcd -advertise-client-urls=http://etcd:2379 -listen-client-urls http://0.0.0.0:2379 --data-dir /etcd minio: container_name: milvus-minio image: minio/minio:RELEASE.2023-03-20T20-16-18Z environment: MINIO_ACCESS_KEY: minioadmin MINIO_SECRET_KEY: minioadmin volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/minio:/minio_data command: minio server /minio_data healthcheck: test: ["CMD", "curl", "-f", "http://localhost:9000/minio/health/live"] interval: 30s timeout: 20s retries: 3 standalone: container_name: milvus-standalone image: milvusdb/milvus:v2.4.4 command: ["milvus", "run", "standalone"] environment: ETCD_ENDPOINTS: etcd:2379 MINIO_ADDRESS: minio:9000 ports: - "19530:19530" - "9091:9091" volumes: - ${DOCKER_VOLUME_DIRECTORY:-.}/volumes/milvus:/var/lib/milvus depends_on: - etcd - minio

把这段保存为docker-compose.yml，然后在同目录执行docker compose up -d。等几个容器都变成 healthy 状态后，再检查一下端口：

docker compose ps

正常会看到 standalone 容器监听 19530 端口。Milvus 的客户端默认用 19530 做 gRPC 通信，9091 是健康检查端口。启动失败时，重点看docker compose logs standalone的输出，大部分问题都出在 etcd 或 MinIO 没起来导致依赖失败。

4.3 创建 Collection：字段、主键、索引

连上 Milvus 之后，第一步是定义 collection 结构。一个 collection 就像关系数据库里的表，每个字段都要明确类型。我在这里存的是记忆片段，所以至少需要这几个字段：

• id：主键，自增
• user_id：用户标识，用于多租户隔离
• conversation_id：来源会话，方便回溯
• text：原始记忆文本
• embedding：向量字段

使用 pymilvus 创建 collection 的代码如下：

from pymilvus import ( connections, Collection, CollectionSchema, FieldSchema, DataType, utility, ) connections.connect(host="localhost", port="19530") COLLECTION_NAME = "agent_memory" DIM = 1024 # 由 embedding 模型决定 # text-embedding-3-small 是 1536 维 # bge-large-zh-v1.5 是 1024 维 fields = [ FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True), FieldSchema(name="user_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="conversation_id", dtype=DataType.VARCHAR, max_length=64), FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=65535), FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=DIM), ] schema = CollectionSchema(fields=fields, description="Agent memory store") if not utility.has_collection(COLLECTION_NAME): collection = Collection(name=COLLECTION_NAME, schema=schema) index_params = { "index_type": "HNSW", "metric_type": "COSINE", "params": {"M": 16, "efConstruction": 200}, } collection.create_index(field_name="embedding", index_params=index_params)

这里两个参数值得展开。第一，metric_type选了COSINE而不是IP（内积）。如果 embedding 做了归一化，两者结果一样，但 COSINE 的分数对文本长度不敏感，召回结果更稳定，我建议默认用 COSINE。第二，HNSW 的M和efConstruction是精度和索引构建速度的平衡：M越大召回越准，但内存占用越高；efConstruction越大索引构建越慢，但查询精度越高。开发环境用M=16, efConstruction=200足够，数据量上了千万后再调参。

4.4 写入与检索：一段可复用的代码

Collection 创建好之后，接下来就是写入和检索，这是整个 Memory 模块最核心的两条路径。

写入路径：把一段文本做 embedding，然后 insert 进 collection。

from pymilvus import Collection def get_embedding(text: str) -> list[float]: # 这里是示意，实际项目根据自己的 embedding 方案实现 # 可以用 OpenAI SDK，也可以用本地 bge 模型 import openai client = openai.OpenAI() resp = client.embeddings.create( model="text-embedding-3-small", input=text, ) return resp.data[0].embedding def add_memory(user_id: str, conversation_id: str, text: str): collection = Collection("agent_memory") embedding = get_embedding(text) collection.insert([ [user_id], [conversation_id], [text], [embedding], ]) def search_memory(user_id: str, query: str, top_k: int = 5): collection = Collection("agent_memory") collection.load() query_embedding = get_embedding(query) results = collection.search( data=[query_embedding], anns_field="embedding", param={"metric_type": "COSINE", "params": {"ef": 64}}, limit=top_k, expr=f'user_id == "{user_id}"', output_fields=["text", "conversation_id", "user_id"], ) hits = [] for hit in results[0]: if hit.distance < 0.7: # 相关度阈值，低于这个分数视为无关 continue hits.append({ "text": hit.entity.get("text"), "conversation_id": hit.entity.get("conversation_id"), "score": hit.distance, }) return hits

检索路径里有几个细节容易踩坑。

第一个是collection.load()。Collection 创建后默认未加载到内存，必须在 search 前执行 load。但频繁 load/unload 会影响性能，实际项目一般在服务启动时统一 load，不要每条请求都 load 一次。

第二个是expr参数。我用user_id == "xxx"做粗过滤，只搜当前用户的记忆。这比全局检索后过滤效率高得多，也是多租户隔离的基础。注意 expr 里的字符串值要加双引号，而且字段类型是 VARCHAR，值最好做一次转义。

第三个是相似度阈值。Milvus 返回的 distance 在 COSINE 下是余弦相似度，范围在 -1 到 1 之间，越大越相关。0.7 这个阈值是我根据项目数据调出来的经验值，你可以根据自己的场景调整。如果阈值太高，召回结果太少；太低，无关内容会污染 prompt。我习惯先把检索结果打印出来看几轮，再定阈值。

4.5 为什么最终选了 Milvus 而不是其他向量库

在写进 Milvus 之前，我还试过 Chroma 和 FAISS。Chroma 上手极快，Python API 很友好，适合做原型；FAISS 只是一个索引库，没有服务端，需要自己处理持久化和并发。

最终换成 Milvus 的原因有三个。第一，Milvus 是独立服务，多个 Agent 实例可以共享同一个记忆存储，解决了 InMemory 阶段“多实例不共享”的硬伤。第二，它自带标量字段过滤能力，expr表达式可以同时按用户、会话、时间过滤向量检索结果，这正是 Memory 场景里高频需要的。第三，数据量大之后可以平滑扩展到分布式模式，不需要推翻重来。

5. 生产级 Memory 模块的进阶设计

5.1 分层记忆：短期、长期、摘要

到了 Milvus 这一层，存储已经不是问题，真正考验设计的是“怎么把不同层级的记忆组合起来用”。

我的做法是三段式组装：短期上下文 + 长期相关记忆 + 全局摘要。短期上下文来自 InMemory 或者文件里的最近 N 轮，保证对话连贯；长期相关记忆从 Milvus 按当前 query 召回，提供背景知识；全局摘要是对历史对话做定期压缩，模型在每轮对话前读一遍摘要，对用户的整体情况有基本感知。

这三段拼进 prompt 后，要控制总量。我的经验是短期上下文控制在 2000 token 以内，召回记忆控制在 3 到 5 条，摘要控制在 500 token 以内。总量超过阈值时，优先砍摘要，再砍短期上下文，最后才砍召回记忆。

5.2 召回不是越多越好

新手最容易犯的错误是“召回的条数越多越好”。实际不是这样。

从 Milvus 里召回 20 条记忆，里面可能有 5 条是高度相关的，剩下 15 条是勉强沾边的。这些勉强沾边的记忆塞进 prompt，不会让模型变聪明，反而会让它分心，甚至把模型带到错误的方向上。

我现在的策略是：限制top_k = 5，并且只保留distance > 0.7的结果。同时做一个“来源去重”：同一个 conversation_id 下最多取 2 条，避免某一个会话的记忆成堆出现，挤压了其他重要记忆的空间。

5.3 写入时机与改写策略

记忆写入不是“每轮对话都写”，而是要有策略。每轮都写会堆积大量低价值信息，比如“好的”“嗯嗯”这种废话，占向量库空间不说，还会降低召回精度。

我的写入规则是：

• 用户显式要求记住时，比如“记住我周一要开会”，必须同步写入；
• LLM 完成了关键任务节点时，比如查完一家公司资料后给出结论，把结论写入；
• 普通轮次对话，用异步方式批量写入，并且存储前先用 LLM 做一次“记忆提炼”，把一段冗长的对话改写成一两句简洁的陈述。

这个“改写后存储”的步骤非常重要。直接用原文存，召回时经常匹配到无意义的重复；改成摘要之后，召回精度和 prompt 的可用性都会明显提升。

5.4 多租户隔离怎么做

只要你的 Agent 服务不止一个人用，就必须考虑隔离。最简单的方案就是在 collection 里加一个user_id或tenant_id字段，写入时带上，检索时在expr里过滤。

但要注意，expr过滤是检索后置条件，不是索引前置条件。也就是说，如果你的向量索引是全库的，哪怕只检索一个用户的数据，复杂度和全库搜索一样。数据量大到一定程度后，更合适的做法是按用户分 shard，或者用 Milvus 的 partition 功能，把不同用户的数据放进不同 partition，检索时指定 partition_names，性能会好很多。

5.5 记忆的更新与遗忘

好的记忆系统不仅要能存能查，还要能“忘”。用户可能换了偏好，可能改了地址，旧记忆就成了干扰项。

Milvus 支持按表达式删除数据：

collection.delete(expr='conversation_id == "abc123"')

我依赖这个能力做定期清理：每 N 天扫描一次写入时间超过 TTL 的会话，删除过期记忆；用户主动要求清除记忆时，按user_id删除全部关联记录。另外要提醒一句，删除操作和插入操作一样，在批量执行时最好做节流，避免高频删除引发 compaction 压力。

6. 常见问题与排查技巧实录

6.1 进程异常退出：0xc0000005 内存访问冲突

如果你在 Windows 本地开发环境跑 Agent，偶尔会遇到进程直接退出，报错：process exited with code 3221225477 (0xC0000005 - memory access violation)

这个错误本质是内存访问越界，常见于底层 C/C++ 扩展库，比如某些本地向量索引库、或者特定版本的 Milvus 客户端依赖。我自己遇到过一次，排查路径是这样的：先看崩溃是否稳定复现，如果只在某个特定操作后出现，多半是传入的数据有问题（比如 embedding 维度与 collection 定义不一致）；如果不稳定复现，优先升级 pymilvus 和依赖库版本，或者换一台 Linux 环境验证。Windows 下 run Docker 的 Milvus 服务端本身一般没问题，问题通常出在本地客户端库的兼容性上。

6.2 连接不上 Milvus 服务

最常见的问题是容器起来了，但客户端连不上 19530 端口。先检查监听状态：

netstat -an | findstr 19530

如果端口没监听，大概率是容器崩了。然后：

docker compose logs standalone

我踩过的一个坑是 Docker 磁盘空间不足导致 Milvus 写入元数据失败，容器反复重启。清理 Docker 磁盘占用后一切正常。另一个坑是本地 19530 端口被其他程序占用，把 compose 里的端口映射改成19531:19530可以快速绕过。

6.3 召回质量差：搜出来一堆不相关的内容

很多人一上来就怀疑 Milvus 有问题，其实大部分时候是 embedding 或者数据预处理的问题。

排查顺序我建议是：第一步，把 query 的 embedding 和召回结果的 embedding 拿出来，用余弦公式手算一遍相似度，确认 Milvus 返回的结果不是错的；第二步，检查存储时是不是整段长文本直接 embedding，长文本语义容易被稀释，最好按 200 到 500 字切块再分别存储；第三步，看 query 改写，用户的原话常常缺少关键信息，先让 LLM 把 query 改写成一个更完整的检索语句，召回效果会有明显提升。

6.4 批量写入慢得离谱

一条一条 insert 肯定慢。Milvus 的写入是按 batch 设计的，一次 insert 几千条和几千次 insert 每次一条的性能差距是两个数量级。

我的做法是先把记忆记录攒在内存里，积累到 100 条或者 1 秒钟后，统一做一次批量插入。批量插入时同时注意 embedding 接口的限流，很多 embedding API 是按每分钟请求数限制的，本地 bge 模型则要注意显存吃紧的问题。

6.5 索引构建失败或查询变慢

索引构建失败最常见的原因是内存不足。HNSW 构建时需要把所有向量加载到内存，数据量大的机器内存不够就会出现 OOM。解法是调小M和efConstruction，或者分批构建。

查询变慢则要看expr过滤条件是否把候选集压得太小，以及 collection 是否频繁 load/unload。还有一个容易忽略的点：collection 里的 segment 数量过多会导致查询变慢，定期执行collection.compact()合并 segment 能改善查询性能。

我自己现在做 Agent 项目，默认路径已经固定下来：先用 InMemory 把流程跑通，再用文件/SQLite 做持久化兜底，等真正需要跨会话语义召回时，再上 Milvus。这个顺序看着土，但每一步都能帮你更清楚地认识到当前项目的瓶颈在哪儿。给 Agent 装上记忆不是一步到位的事，而是一个从“能跑”到“跑得好”的渐进过程。