企业官网建设流程全解析

1. 从“湖生万物”说起：这个平台到底在解决什么问题

第一次看到“湖生万物，助力 AI”这个提法，我脑子里冒出来的不是诗意的画面，而是一堆工程上的麻烦事。做过 Agent 项目的人都知道，一个能跑起来的智能体，背后往往挂着七八个数据源：对话历史、工具调用日志、向量库、结构化业务表、图片、音频、视频，甚至还有实时传感器流。这些东西格式各异、更新频率不同、生命周期也不一样，传统的数据平台根本兜不住。

“湖生万物”这个说法，我理解它指向的是一种数据湖思路的升级——不是简单地把数据堆在一起，而是让湖里能“长出”各种形态的数据资产，并且这些资产能被 Agent 直接消费。面向 Agent 的全模态数据平台，核心要解决的就是三件事：多模态数据的统一接入与治理、面向 Agent 的低延迟检索与编排、以及数据在 Agent 生命周期中的记忆与演化。

这个内容适合谁来参考？如果你正在做 Agent 开发，被“记忆怎么存”“工具返回的数据怎么管”“多模态输入怎么统一处理”这些问题卡住，那这篇就是写给你的。如果你只是刚听说 Agent 这个概念，也没关系，我会尽量用生活化的类比把底层逻辑讲清楚。全模态数据平台不是某个具体产品，而是一类架构思路，理解它比记住某个工具名字重要得多。

我个人的判断是，2026 年这个时间节点上，Agent 的竞争已经从“模型能力”转向“数据能力”。模型大家都能调用，但你的 Agent 能不能记住三个月前用户说过的一句话、能不能在毫秒级从几百万条多模态记录里找到相关片段、能不能在工具调用失败时优雅降级，这些才是真正拉开差距的地方。数据平台就是干这个的。

2. 全模态数据平台的架构拆解与选型逻辑

2.1 为什么传统数据栈撑不住 Agent 场景

先说说我踩过的坑。早期做 Agent 项目，我试过直接用关系型数据库存对话历史，用文件系统存图片，用独立的向量库存 embedding。跑 demo 没问题，一上量就崩。问题出在三个地方：

第一，数据孤岛导致检索割裂。用户问“上次我发的那个红色图表的分析结论是什么”，你需要同时查对话历史、图片元数据、分析结果表，三个系统各查各的，延迟叠加，而且很难做联合排序。

第二，Schema 僵化。Agent 产生的数据形态变化太快，今天存的是工具调用参数，明天可能要存多轮推理的中间状态。关系型数据库改表结构的速度跟不上 Agent 迭代的速度。

第三，生命周期管理缺失。Agent 的记忆不是永久有效的，有些短期上下文几小时后就该淘汰，有些长期偏好要保留几个月。传统数据栈没有原生的 TTL 和分层存储机制。

全模态数据平台的设计思路，本质上是用湖仓一体的底座 + 多模态索引层 + Agent 语义层来替代这套拼凑方案。湖仓一体解决统一存储和 Schema 演化，多模态索引层解决跨模态检索，Agent 语义层解决记忆管理和上下文编排。

2.2 存储层选型：对象存储打底，开放格式优先

存储层我的建议很明确：对象存储做底座，开放列式格式做表管理。具体来说，S3 兼容的对象存储负责存原始数据，Parquet 或 ORC 做列式存储，Iceberg 或 Hudi 做表格式管理。这套组合的好处是存算分离，Agent 的检索负载和写入负载可以独立扩缩容。

为什么不用纯向量数据库做底座？因为向量库擅长相似度检索，但不擅长范围查询、聚合分析、事务更新。Agent 场景里这两种需求是混合的。比如“找出过去一周所有调用过天气工具且失败率超过 10% 的会话”，这是典型的分析型查询，向量库做不了。

为什么不用纯关系型数据库？前面说过了，Schema 演化和多模态支持是硬伤。当然，如果你的 Agent 场景非常简单，只有结构化数据，那关系型数据库依然是好选择。选型要看场景，不要为了“全模态”而全模态。

提示：对象存储的选型要注意小文件问题。Agent 产生的日志和中间状态往往是小文件，直接写对象存储会导致元数据压力过大。常见做法是先写本地 WAL，攒批后合并上传，或者用支持小文件合并的表格式。

2.3 多模态索引层：统一 Embedding 空间是关键

多模态检索最难的地方在于，文本、图片、音频的向量不在同一个空间里，没法直接算相似度。常见方案有两种：一种是分别建索引，检索时做多路召回再融合排序；另一种是训练统一的跨模态 Embedding 模型，把所有模态映射到同一空间。

第一种方案工程上更可控，我目前更推荐。具体做法是：文本用文本 Embedding 模型，图片用 CLIP 类模型，音频用语音 Embedding 模型，各自建 HNSW 或 IVF 索引。检索时，如果 query 是文本，就分别去文本索引和图片索引里召回，然后用一个轻量级的重排序模型做融合。这个重排序模型可以很简单，比如基于规则打分，也可以是一个小的交叉编码器。

第二种方案理论上更优雅，但训练成本高，而且跨模态对齐的效果在垂直领域往往不如分别建模。除非你的场景对跨模态检索要求极高，否则不建议一上来就搞统一空间。

索引层还有一个容易被忽略的点：元数据过滤。Agent 检索时往往带条件，比如“只查这个用户的数据”“只查最近三天的”。纯向量检索不支持这种过滤，需要在索引结构里嵌入元数据字段，或者用前置过滤 + 后置过滤的组合。我实测下来，前置过滤（先按元数据缩小候选集再算向量）在候选集不大时效率更高，后置过滤适合候选集大的场景。

2.4 Agent 语义层：记忆分级与上下文编排

这一层是全模态数据平台区别于普通数据平台的核心。Agent 的记忆不是单一维度的，我习惯把它分成四级：

• 瞬时记忆：当前对话轮次的上下文，存在内存或 Redis，TTL 几分钟到几小时。
• 工作记忆：当前任务相关的工具调用结果、中间推理状态，存在高速存储，TTL 几小时到几天。
• 长期记忆：用户偏好、历史决策、重要事实，存在持久化存储，需要定期压缩和摘要。
• 归档记忆：冷数据，用于审计和回溯，存在低成本存储，检索频率极低。

上下文编排要解决的是：给定当前 query，从这四级记忆里各取多少、怎么拼、怎么控制 token 预算。我的经验是，瞬时记忆全量保留，工作记忆按相关性取 top-k，长期记忆做摘要后取少量高置信度条目，归档记忆基本不参与实时推理。

注意：长期记忆的摘要策略很关键。我试过直接截断，效果很差；后来改成用一个小模型做增量摘要，每积累 N 轮对话就更新一次摘要，检索时用摘要而非原文，token 消耗降了 60% 以上，效果基本没损失。

3. 核心实操：从零搭建一个最小可用的全模态数据平台

3.1 环境准备与依赖安装

假设你已经有基本的 Python 和 Docker 环境，下面是我在本地验证过的一套最小配置。这套配置不追求生产级性能，但能让你把整个链路跑通，理解每个环节在干什么。

# 创建虚拟环境 python -m venv agent-data-platform source agent-data-platform/bin/activate # 核心依赖 pip install fastapi uvicorn pip install minio # 对象存储客户端 pip install pyiceberg # 表格式管理 pip install sentence-transformers # 文本 Embedding pip install open-clip-torch # 图片 Embedding pip install faiss-cpu # 向量索引 pip install redis # 瞬时记忆 pip install sqlalchemy # 元数据管理

对象存储我用 MinIO 本地起一个，生产环境换成任意 S3 兼容服务即可。Iceberg 的 catalog 可以用 SQLite 做本地元数据存储，生产环境换成 Hive Metastore 或 REST Catalog。

# 启动 MinIO docker run -d -p 9000:9000 -p 9001:9001 \ -e MINIO_ROOT_USER=admin \ -e MINIO_ROOT_PASSWORD=password \ minio/minio server /data --console-address ":9001"

这套环境跑起来大概占 2GB 内存，普通开发机完全够用。如果你要处理视频，需要额外装 ffmpeg 做抽帧，内存建议加到 8GB 以上。

3.2 多模态数据接入管道实现

数据接入管道要解决的是：不同来源的数据进来后，怎么统一成平台能理解的格式。我的做法是定义一个统一数据单元（Unified Data Unit），所有模态的数据都转成这个结构再入库。

from dataclasses import dataclass, field from typing import Any, Optional import time import uuid @dataclass class UnifiedDataUnit: id: str = field(default_factory=lambda: str(uuid.uuid4())) modality: str = "text" # text, image, audio, video, structured content: Any = None # 原始内容或引用路径 embedding: Optional[list] = None metadata: dict = field(default_factory=dict) timestamp: float = field(default_factory=time.time) ttl: Optional[int] = None # 秒，None 表示永久 source: str = "unknown" agent_id: Optional[str] = None session_id: Optional[str] = None

这个结构看起来简单，但每个字段都有讲究。modality决定后续走哪个索引管道；content对于大文件存路径而非内容本身；metadata放业务相关的过滤字段；ttl驱动生命周期管理；agent_id和session_id是 Agent 场景特有的，用于隔离不同 Agent 和会话的数据。

文本接入相对简单，直接调 Embedding 模型。图片接入需要先做预处理：统一分辨率、去重、可选地做 OCR 提取文字。音频接入要先做语音转文字，再把文字和原始音频都存下来。视频接入最复杂，我通常抽关键帧做图片处理，音频轨做语音处理，然后通过时间戳关联。

from sentence_transformers import SentenceTransformer text_model = SentenceTransformer('all-MiniLM-L6-v2') def ingest_text(text: str, metadata: dict, agent_id: str, session_id: str): embedding = text_model.encode(text).tolist() unit = UnifiedDataUnit( modality="text", content=text, embedding=embedding, metadata=metadata, agent_id=agent_id, session_id=session_id ) return unit

这里有个实操细节：Embedding 模型的选择要跟你的检索场景匹配。all-MiniLM-L6-v2只有 384 维，速度快但精度一般，适合做原型验证。生产环境如果追求精度，可以换成更大的模型，但要注意推理延迟和存储成本。我一般会准备两个模型，小模型做粗排，大模型做精排。

3.3 向量索引构建与混合检索

索引构建我推荐用 FAISS，原因是它足够轻量，而且支持多种索引类型。对于百万级以下的数据量，HNSW 索引在召回率和延迟之间平衡得最好。

import faiss import numpy as np class VectorIndex: def __init__(self, dim: int): self.dim = dim self.index = faiss.IndexHNSWFlat(dim, 32) self.index.hnsw.efConstruction = 200 self.id_map = {} # faiss_id -> unit_id self.next_id = 0 def add(self, unit_id: str, embedding: list): vec = np.array([embedding], dtype='float32') self.index.add(vec) self.id_map[self.next_id] = unit_id self.next_id += 1 def search(self, query_embedding: list, top_k: int = 10): vec = np.array([query_embedding], dtype='float32') distances, indices = self.index.search(vec, top_k) results = [] for dist, idx in zip(distances[0], indices[0]): if idx in self.id_map: results.append((self.id_map[idx], float(dist))) return results

efConstruction这个参数控制建索引时的搜索深度，值越大索引质量越高但建索引越慢。200 是我常用的折中值。检索时的efSearch参数可以在查询时动态调整，追求召回就调大，追求延迟就调小。

混合检索的关键在于多路召回 + 融合排序。文本 query 同时去文本索引和图片索引召回，然后按加权分数合并。权重可以根据场景调，比如客服场景文本权重大，电商场景图片权重大。

def hybrid_search(query: str, text_index, image_index, top_k=10, text_weight=0.7): text_emb = text_model.encode(query).tolist() text_results = text_index.search(text_emb, top_k * 2) # 图片索引用同一个文本 Embedding 做跨模态检索 image_results = image_index.search(text_emb, top_k * 2) # 归一化分数后加权融合 merged = {} for uid, score in text_results: merged[uid] = merged.get(uid, 0) + text_weight * (1 - score) for uid, score in image_results: merged[uid] = merged.get(uid, 0) + (1 - text_weight) * (1 - score) return sorted(merged.items(), key=lambda x: x[1], reverse=True)[:top_k]

提示：跨模态检索时，图片索引的 Embedding 模型必须和文本 Embedding 模型在同一个语义空间，否则检索结果没有意义。CLIP 类模型天然满足这个条件，这也是我推荐用 CLIP 做图片 Embedding 的原因。

3.4 记忆分级存储与上下文组装

记忆分级存储的实现，核心是根据 TTL 和访问频率做数据分层。我用 Redis 存瞬时记忆和工作记忆，用 Iceberg 表存长期记忆和归档记忆。

import redis import json r = redis.Redis(host='localhost', port=6379, decode_responses=True) def store_short_term(session_id: str, unit: UnifiedDataUnit, ttl: int = 3600): key = f"stm:{session_id}:{unit.id}" r.setex(key, ttl, json.dumps({ "content": unit.content, "metadata": unit.metadata, "timestamp": unit.timestamp })) def get_recent_context(session_id: str, limit: int = 20): pattern = f"stm:{session_id}:*" keys = r.keys(pattern) items = [] for key in keys: data = json.loads(r.get(key)) items.append(data) items.sort(key=lambda x: x["timestamp"], reverse=True) return items[:limit]

上下文组装是 Agent 语义层最核心的逻辑。我的做法是写一个ContextAssembler类，输入当前 query 和 session_id，输出拼好的上下文。

class ContextAssembler: def __init__(self, text_index, long_term_store, token_budget=4000): self.text_index = text_index self.long_term_store = long_term_store self.token_budget = token_budget def assemble(self, query: str, session_id: str): # 1. 瞬时记忆：最近 N 轮对话 recent = get_recent_context(session_id, limit=10) # 2. 工作记忆：当前任务相关的工具调用结果 working = self._get_working_memory(session_id) # 3. 长期记忆：向量检索相关历史 query_emb = text_model.encode(query).tolist() long_term = self.text_index.search(query_emb, top_k=5) # 4. 按 token 预算组装 context = self._pack(recent, working, long_term) return context def _pack(self, recent, working, long_term): # 优先级：瞬时 > 工作 > 长期 # 超出预算时从低优先级开始截断 packed = [] used = 0 for item in recent + working + long_term: tokens = self._count_tokens(item) if used + tokens > self.token_budget: break packed.append(item) used += tokens return packed

这个组装逻辑看起来简单，但实际调优空间很大。比如长期记忆的检索结果要不要做摘要、工作记忆要不要按工具类型分组、瞬时记忆要不要做滑动窗口，这些都会影响最终效果。我的经验是先用最简单的版本跑通，然后根据 bad case 逐步调优。

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

4.1 检索结果不相关：从 Embedding 到索引逐层排查

这是最高频的问题。用户 query 和检索结果对不上，原因可能出在四个地方：Embedding 模型不匹配、索引参数不合理、元数据过滤过严、或者 query 本身有歧义。

我的排查顺序是：先看 query 的 Embedding 和 top-1 结果的 Embedding 余弦相似度，如果低于 0.5，基本是模型问题；如果相似度正常但结果不对，检查索引的efSearch参数是不是太小；如果参数正常，检查元数据过滤条件是不是把正确结果排除了；最后才怀疑 query 歧义，这时候需要做 query 改写或扩展。

注意：Embedding 模型微调是最后手段，成本高周期长。先确认是不是索引和过滤的问题，这两类问题解决起来快得多。

4.2 写入延迟高：小文件合并与批量提交

Agent 场景的写入特点是高频小批量。每次工具调用、每轮对话都产生一条记录，如果每条都单独写对象存储，元数据压力会非常大。我实测过，单条写入的延迟是批量写入的 10 倍以上。

解决方案是本地缓冲 + 批量提交。在 Agent 进程内维护一个写入队列，攒到一定条数或一定时间后批量提交。批量大小需要权衡：太小起不到合并效果，太大增加内存压力和丢失风险。我的经验值是 100 条或 5 秒，取先到者。

import threading import time class BatchWriter: def __init__(self, flush_interval=5, batch_size=100): self.buffer = [] self.flush_interval = flush_interval self.batch_size = batch_size self.lock = threading.Lock() self._start_flush_thread() def write(self, unit): with self.lock: self.buffer.append(unit) if len(self.buffer) >= self.batch_size: self._flush() def _flush(self): if not self.buffer: return batch = self.buffer[:] self.buffer = [] # 实际写入逻辑 self._do_write(batch) def _start_flush_thread(self): def loop(): while True: time.sleep(self.flush_interval) with self.lock: self._flush() t = threading.Thread(target=loop, daemon=True) t.start()

这个模式在 Agent 进程重启时会丢失缓冲区数据，生产环境需要配合 WAL 做持久化。但对于大多数场景，丢失最后几秒的数据是可以接受的，换来的是写入吞吐量的大幅提升。

4.3 记忆膨胀：TTL 策略与摘要压缩

跑一段时间后，长期记忆会膨胀到不可控。我见过一个客服 Agent 跑了三个月，长期记忆表涨到 2TB，检索延迟从 50ms 涨到 2s。问题出在没有做记忆淘汰和压缩。

TTL 策略要按记忆类型分别设置。瞬时记忆 TTL 设 1 小时，工作记忆设 24 小时，长期记忆设 90 天，归档记忆永久但迁移到低成本存储。这个 TTL 不是拍脑袋定的，要根据业务场景调。比如金融场景的合规要求可能要求长期记忆保留更久，而闲聊场景的记忆可能一周就够了。

摘要压缩是另一个关键手段。我的做法是：长期记忆每积累 50 条同主题记录，触发一次摘要生成，用一个小模型把 50 条压缩成 1 条摘要，原始记录迁移到归档存储。检索时优先命中摘要，需要细节时再回查归档。

def compress_memory(units: list, summarizer): texts = [u.content for u in units if u.modality == "text"] combined = "\n".join(texts) summary = summarizer(combined, max_length=200) return UnifiedDataUnit( modality="text", content=summary, metadata={"type": "summary", "source_count": len(units)}, ttl=90 * 24 * 3600 )

实测下来，摘要压缩能把长期记忆的存储量降低 80% 以上，检索延迟降低 60%，而关键信息的召回率损失不到 5%。这个投入产出比非常划算。

4.4 Agent 执行中断后的数据一致性

Agent 执行过程中可能因为各种原因中断，这时候数据一致性就成了问题。比如工具调用已经写入结果，但 Agent 状态没更新，下次恢复时就会重复调用。

我的解决方案是两阶段提交 + 幂等键。每个 Agent 任务分配一个唯一的 task_id，所有写入操作都带上这个 task_id 作为幂等键。写入前先检查该 task_id 是否已有记录，有则跳过。这样即使中断后重试，也不会产生重复数据。

def idempotent_write(unit, task_id, store): existing = store.query_by_task_id(task_id) if existing: return existing unit.metadata["task_id"] = task_id store.write(unit) return unit

这个机制看起来简单，但能避免大量脏数据问题。我踩过的坑是：早期没做幂等，Agent 重试导致同一条工具调用结果写了三遍，检索时返回重复内容，用户体验很差。加上幂等键之后，这类问题基本消失了。

5. 从原型到生产：性能调优与扩展性考量

5.1 索引分片与水平扩展

单机 FAISS 索引在数据量超过千万级后，内存和检索延迟都会成为瓶颈。这时候需要做索引分片。分片策略有两种：按数据量均匀分片和按业务维度分片。我推荐按业务维度分，比如按 agent_id 或 tenant_id 分片，这样检索时可以只查相关分片，减少跨分片查询。

分片后的检索需要做结果合并。如果各分片返回的分数可比，直接归并排序即可；如果不可比，需要做分数归一化。我通常用 min-max 归一化，把各分片的分数映射到 [0,1] 区间再合并。

def sharded_search(query_emb, shards, top_k=10): all_results = [] for shard in shards: results = shard.search(query_emb, top_k) if results: scores = [s for _, s in results] min_s, max_s = min(scores), max(scores) for uid, score in results: norm = (score - min_s) / (max_s - min_s + 1e-8) all_results.append((uid, norm)) all_results.sort(key=lambda x: x[1], reverse=True) return all_results[:top_k]

分片的另一个好处是支持滚动更新。重建索引时不需要全量重建，可以逐分片更新，服务不中断。

5.2 缓存策略：多级缓存降低检索延迟

Agent 场景的检索有明显的热点集中特征。少数高频 query 占了大部分检索量。这时候加缓存收益很高。我用两级缓存：本地 LRU 缓存 + Redis 分布式缓存。

本地缓存存最热的 query，容量小但延迟极低（微秒级）。Redis 缓存存次热 query，容量大但延迟稍高（毫秒级）。缓存 key 用 query 的哈希，value 存检索结果 ID 列表。缓存失效策略用 TTL + 主动失效结合，数据更新时主动清除相关缓存。

from functools import lru_cache import hashlib @lru_cache(maxsize=1000) def cached_search(query_hash: str): # 先查 Redis cached = r.get(f"search:{query_hash}") if cached: return json.loads(cached) # 缓存未命中，实际检索 results = do_search(query_hash) r.setex(f"search:{query_hash}", 300, json.dumps(results)) return results

实测下来，加缓存后 P99 延迟从 200ms 降到 30ms，效果非常明显。但要注意缓存一致性，数据更新频繁的场景要缩短 TTL 或做主动失效。

5.3 监控指标与告警设置

生产环境必须有的监控指标：检索延迟（P50/P95/P99）、召回率、写入吞吐量、存储增长率、缓存命中率、错误率。这些指标我建议用 Prometheus + Grafana 做可视化，告警阈值根据业务 SLA 定。

我特别想强调的是召回率监控。这个指标不像延迟那么直观，但对 Agent 效果影响最大。我的做法是定期用标注好的 query-结果对做离线评估，召回率下降超过 5% 就告警。线上则用点击率或用户反馈做代理指标。

这套监控跑起来后，大部分问题都能在用户感知之前发现。我踩过的坑是早期没做召回率监控，等用户反馈“Agent 变笨了”才去查，发现是索引参数被误改了，白白损失了几天效果。

6. 几个容易被忽略的工程细节

6.1 多租户隔离与数据安全

如果你的平台要服务多个 Agent 或多个用户，隔离是必须的。隔离有三个层次：存储隔离、索引隔离、检索隔离。存储隔离用不同的 bucket 或前缀，索引隔离用不同的索引实例，检索隔离在查询时强制加 tenant_id 过滤。

我推荐至少做到索引隔离 + 检索隔离。存储隔离成本高，除非有强合规要求，否则共享存储 + 逻辑隔离就够了。检索隔离是底线，绝对不能省，否则会出现跨租户数据泄露。

def tenant_search(query_emb, tenant_id, index): # 强制加租户过滤 results = index.search(query_emb, top_k=50) filtered = [] for uid, score in results: unit = get_unit(uid) if unit.metadata.get("tenant_id") == tenant_id: filtered.append((uid, score)) return filtered[:10]

这个过滤逻辑看起来简单，但一定要在检索层强制做，不能依赖上层调用方传参。我见过因为上层忘了传 tenant_id 导致的数据泄露事故，教训很深刻。

6.2 Schema 演化与向后兼容

Agent 产生的数据形态变化快，Schema 演化是常态。Iceberg 支持 Schema 演化，加列、改列类型、重命名都可以在线做。但要注意向后兼容：新代码要能读旧数据，旧代码要能读新数据（至少不报错）。

我的做法是：新增字段一律给默认值，删除字段先标记废弃而不是直接删，改类型只做兼容性扩展（如 int 改 long）。这样新旧代码可以共存，滚动升级不会出问题。

提示：Schema 演化后要重建索引，否则新字段的过滤条件不生效。重建索引可以后台异步做，不影响线上检索。

6.3 冷热数据分层与成本控制

全模态数据平台的存储成本很容易失控，尤其是图片和视频。冷热分层是必须的。热数据（最近 7 天）放高速存储，温数据（7-90 天）放标准对象存储，冷数据（90 天以上）放归档存储。

分层迁移可以按 TTL 自动触发，也可以用访问频率驱动。我通常两个策略结合：TTL 保证冷数据一定迁移，访问频率保证热数据不被误迁。迁移过程要异步做，避免影响线上检索。

成本控制还有一个手段是数据去重。Agent 场景里重复数据很多，比如同一个工具被反复调用返回相同结果。做内容哈希去重能省不少存储。我实测过去重率在 20%-40% 之间，取决于场景。

7. 我个人的一些实践体会

这套全模态数据平台我从原型到生产跑了大概半年，最大的体会是：不要追求一步到位。一开始就想做统一 Embedding 空间、做智能摘要、做自动分层，结果每个环节都出问题，排查起来极其痛苦。后来改成先用最简单的方案跑通链路，再逐个环节优化，反而进展更快。

另一个体会是监控比功能重要。功能可以慢慢加，但监控必须一开始就有。没有监控，你根本不知道系统在发生什么，出了问题只能靠猜。我现在的习惯是，任何新模块上线前，先把监控指标和告警配好，再考虑功能。

最后分享一个小技巧：用真实 Agent 流量做回归测试。我每周会抽一批线上真实 query，跑一遍检索，人工评估 top-10 结果的相关性。这个习惯帮我发现了不少离线指标看不出来的问题，比如某些长尾 query 的检索效果特别差，但在平均指标里被掩盖了。

这个方向后续还可以扩展的地方很多，比如多 Agent 协作时的数据共享机制、Agent 记忆的联邦学习、跨平台的数据互操作等。但那是另一个话题了，先把单 Agent 的数据平台做扎实，比什么都重要。