智能体数据不是日志越多越好：用 TypeScript 构建轨迹清洗与质量分层流水线

智能体日志的价值不取决于数量，而取决于轨迹是否完整、可比较、可复现。本文用 TypeScript 实现一条从原始日志到版本化数据集的流水线，覆盖清洗、脱敏、去重、质量评分和分层导出。这套流程适合微调、提示词优化和离线评测，但不会把一个分数包装成绝对真相，所有筛选依据都应保留为可解释字段。

先定义轨迹，而不是直接收集日志

智能体的一次运行通常包含用户输入、模型多轮调用、工具调用、工具返回、最终答案和运行状态。工程上最容易犯的错误，是把每一行日志都当成训练样本。这样会带来几个问题：同一任务的中间步骤被重复计算，失败重试占据大量比例，工具异常被误当成模型能力问题，甚至把包含密钥和个人信息的原始内容直接送入下游系统。

本文将一条可用于分析的轨迹定义为一个整体对象，而不是某一条事件：

trace = input + ordered events + final output + outcome + metadata

其中 events 必须保留顺序，outcome 需要区分成功、失败、超时和人工拒绝。不要只用 final_output 是否为空来判断成功，因为有些工具调用失败后，模型仍然会生成一个看似完整但事实错误的回答。

推荐先建立最小字段契约：

字段	用途	质量风险
`traceId`	关联一次运行	重试时可能重复或缺失
`input`	用户任务和上下文	可能含个人信息
`events`	模型与工具交互序列	事件顺序错乱会改变语义
`output`	最终回答	不能单独代表任务成功
`status`	运行结果	需要统一枚举值
`metadata`	模型、版本、耗时等信息	字段漂移会影响比较

清洗、脱敏和结构校验

清洗顺序会影响结果。先做结构校验，再做内容规范化和脱敏，最后计算指纹；如果在脱敏前计算哈希，相同用户信息会导致样本无法有效聚合，也会增加敏感内容进入索引的风险。

下面的示例只依赖 Node.js 内置模块。将代码保存为 pipeline.ts，使用 npm install -D typescript tsx 后运行 npx tsx pipeline.ts。它演示了一个可运行的单文件版本：读取内存中的原始轨迹，过滤非法记录，脱敏，规范化文本，计算去重指纹，并输出质量分层结果。

import { createHash } from "node:crypto";

type Status = "success" | "failed" | "timeout" | "rejected";
type EventKind = "model" | "tool" | "tool_result";

interface RawEvent {
  kind: EventKind;
  name?: string;
  content?: string;
}

interface RawTrace {
  traceId: string;
  input?: string;
  events?: RawEvent[];
  output?: string;
  status?: string;
  model?: string;
}

interface CleanTrace {
  traceId: string;
  input: string;
  events: RawEvent[];
  output: string;
  status: Status;
  model: string;
  fingerprint: string;
  quality: number;
  tier: "gold" | "silver" | "quarantine";
  reasons: string[];
}

const validStatuses = new Set<Status>(["success", "failed", "timeout", "rejected"]);

function redact(text: string): string {
  return text
    .replace(/sk-[A-Za-z0-9_-]{10,}/g, "[REDACTED_API_KEY]")
    .replace(/[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Za-z]{2,}/g, "[REDACTED_EMAIL]")
    .replace(/\\b1[3-9]\\d{9}\\b/g, "[REDACTED_PHONE]");
}

function normalize(text: string): string {
  return redact(text).normalize("NFKC").replace(/\\s+/g, " ").trim().toLowerCase();
}

function fingerprint(trace: Pick<CleanTrace, "input" | "events" | "output">): string {
  const body = JSON.stringify({
    input: normalize(trace.input),
    events: trace.events.map((event) => ({
      kind: event.kind,
      name: event.name ?? "",
      content: normalize(event.content ?? "")
    })),
    output: normalize(trace.output)
  });
  return createHash("sha256").update(body).digest("hex");
}

function clean(raw: RawTrace): CleanTrace | null {
  if (!raw.traceId || !raw.input || !raw.output || !raw.events?.length) return null;
  if (!raw.status || !validStatuses.has(raw.status as Status)) return null;

  const trace = {
    traceId: raw.traceId,
    input: redact(raw.input),
    events: raw.events.map((event) => ({
      kind: event.kind,
      name: event.name ? normalize(event.name) : undefined,
      content: event.content ? redact(event.content) : undefined
    })),
    output: redact(raw.output),
    status: raw.status as Status,
    model: raw.model ?? "unknown"
  };

  const reasons: string[] = [];
  let quality = 100;
  if (trace.status !== "success") {
    quality -= 45;
    reasons.push(`status:${trace.status}`);
  }
  if (trace.events.some((event) => event.kind === "tool" && !event.name)) {
    quality -= 15;
    reasons.push("tool_without_name");
  }
  if (trace.output.length < 20) {
    quality -= 20;
    reasons.push("short_output");
  }
  quality = Math.max(0, quality);

  const withScore = { ...trace, fingerprint: "", quality, tier: "quarantine" as const, reasons };
  withScore.fingerprint = fingerprint(withScore);
  withScore.tier = quality >= 80 ? "gold" : quality >= 50 ? "silver" : "quarantine";
  return withScore;
}

const rawTraces: RawTrace[] = [
  {
    traceId: "run-001",
    input: "请总结订单状态，联系 alice@example.com",
    events: [
      { kind: "tool", name: "get_order", content: "order_id=1001" },
      { kind: "tool_result", content: "已发货" }
    ],
    output: "订单 1001 已发货，预计明天送达。",
    status: "success",
    model: "model-a"
  },
  {
    traceId: "run-002",
    input: "请总结订单状态，联系 alice@example.com",
    events: [
      { kind: "tool", name: "get_order", content: "order_id=1001" },
      { kind: "tool_result", content: "已发货" }
    ],
    output: "订单 1001 已发货，预计明天送达。",
    status: "success",
    model: "model-a"
  },
  {
    traceId: "run-003",
    input: "查询库存",
    events: [{ kind: "tool", content: "warehouse timeout" }],
    output: "暂时无法查询。",
    status: "timeout",
    model: "model-a"
  }
];

const cleaned = rawTraces.map(clean).filter((trace): trace is CleanTrace => trace !== null);
const unique = [...new Map(cleaned.map((trace) => [trace.fingerprint, trace])).values()];

console.log(JSON.stringify({ total: rawTraces.length, valid: cleaned.length, unique: unique.length, data: unique }, null, 2));

示例中的脱敏规则只是边界保护，不是完整的隐私方案。生产环境还应结合字段级访问控制、日志保留期限和人工抽样检查。对于身份证号、地址、医疗信息等内容，正则表达式很容易漏检，必要时应接入经过评估的实体识别组件，并把脱敏命中情况写入审计字段，而不是覆盖原始数据后无法追踪。

近重复检测与失败轨迹隔离

精确哈希只能识别规范化后完全相同的轨迹。实际数据中常见的是近重复：输入只多了时间戳，工具结果多了一段请求 ID，或者模型答案只改变了标点。可以先删除明显的动态字段，再采用 token 集合相似度、编辑距离或向量检索做候选发现。

不要一开始就对全量数据做昂贵的两两比较。较稳妥的处理方式是先按任务类型、工具集合和模型版本分桶，再在桶内比较。对于每个近重复集合保留一个代表样本，代表选择应有明确规则，例如优先选择成功状态、事件更完整、耗时字段齐全且质量分更高的轨迹。被合并的样本不能直接丢弃，应保留 duplicateOf、duplicateMethod 和 duplicateScore，这样后续可以解释样本数量为什么变化。

失败轨迹不等于无价值。它们可以用于故障分析、拒答评测和恢复策略训练，但不应和成功示范样本混在同一个微调集合中。建议至少拆成三层：

分层	典型条件	推荐用途
`gold`	成功、事件完整、无明显敏感信息、通过人工或规则检查	高置信微调和核心离线评测
`silver`	基本可用，但存在轻微缺失或只通过自动检查	提示词实验、候选训练集
`quarantine`	失败、超时、结构异常、隐私风险或重复关系不明确	故障分析和人工复核

这里的分层是数据治理策略，不是模型能力评级。一个成功轨迹也可能因为答案事实错误而进入隔离区，因此质量评分最好增加任务专用检查器，例如 JSON Schema 校验、SQL 执行验证或基于参考答案的关键字段比对。

让质量评分可解释

一个简单、可审计的分数可以从 100 分开始扣分，而不是直接训练一个黑盒质量模型。基础维度包括：状态是否成功、输入输出是否完整、事件链是否闭合、工具调用是否有对应返回、是否存在异常堆栈或敏感信息、答案长度是否合理。

分数必须与原因同时保存。例如 status:timeout、tool_without_name 和 short_output 比单独的 quality=40 更有用，因为数据工程师可以据此修正规则，也能在版本升级时解释样本数量的变化。评分函数还应区分硬规则和软规则：发现 API 密钥可以直接隔离，输出稍短则只扣分。不要把所有异常都折算成一个不可逆的数字。

对于结果偏置，需要额外记录任务和结果分布。只保留成功轨迹会产生明显的幸存者偏差，模型看不到工具失败、权限不足和用户拒绝等现实情况。可以按任务类型、状态、工具、模型版本和难度分组统计，然后设置采样上限或分层抽样。这样既避免某个高频简单任务淹没数据，也不会为了追求均衡而人为制造不存在的样本。

数据版本与验证清单

清洗结果应是不可变的数据集产物，而不是覆盖原日志。一个版本目录可以包含 traces.jsonl、manifest.json 和 quality-report.json。清单至少记录原始输入数量、通过校验数量、去重数量、各分层数量、规则版本、代码提交标识和生成时间。原始日志只允许受控访问，导出的训练集则只包含必要字段。

例如：

{
  "dataset": "agent-traces",
  "version": "2025-03-08.rules-v2",
  "sourceCount": 12000,
  "validCount": 10980,
  "uniqueCount": 8420,
  "tiers": { "gold": 5100, "silver": 2200, "quarantine": 1120 },
  "rulesVersion": "2.1.0",
  "schemaVersion": "1.0.0"
}

发布前可以用固定的验证清单：确认 JSONL 每行都能解析；所有轨迹都包含 schema 版本；指纹在同一版本内唯一；脱敏规则对测试样本全部命中；工具调用和工具返回的数量关系合理；每个分层都保留抽样样本；版本之间的数量变化超过阈值时阻止自动发布。还应建立少量“金标准”样本，由人工确认预期状态和质量等级，作为规则回归测试。

后续的微调、提示词优化和离线评测应分别取数。微调偏向高质量、多样性的成功轨迹；提示词优化可以同时使用成功和失败案例，但必须标注失败原因；离线评测需要固定测试集，不能每次从最新日志中随机抽取，否则指标会随着数据清洗规则一起漂移。数据版本和实验版本要关联起来，才能回答“这次效果变化来自提示词、模型，还是数据集变化”。

总结

智能体数据治理的重点不是把日志收集得更多，而是让每条轨迹具备明确的结构、状态和使用边界。实践中可以按以下顺序落地：

定义轨迹 schema，区分运行状态、事件序列和最终答案。
先校验结构，再规范化和脱敏，最后计算内容指纹。
通过分桶和近重复检测减少样本冗余，并保留合并关系。
用可解释的规则评分，把 gold、silver 和 quarantine 分开使用。
对失败样本和结果分布做单独分析，避免只保留成功案例造成偏置。
输出不可变的数据版本、清单和质量报告，为实验结果提供可追溯依据。

这条流水线不替代人工判断，也不保证数据天然正确。它的价值在于把隐含的筛选标准变成代码、字段和版本，让团队可以复查每一次数据变化，并把更多精力放在任务定义和模型行为本身。

智能体数据不是日志越多越好：用 TypeScript 构建轨迹清洗与质量分层流水线

先定义轨迹，而不是直接收集日志

清洗、脱敏和结构校验

近重复检测与失败轨迹隔离

让质量评分可解释

数据版本与验证清单

总结

相关文章

实时语音智能体状态机：处理打断、轮次与音频背压

别让模型差异渗进业务代码：TypeScript LLM Provider 适配层实战

多智能体不是多开模型：用 TypeScript 构建可终止任务图

评论