上线即平台期：拆开 DigitalOcean 的 Agent 学习服务与数据服务

DigitalOcean 于 2026 年 10 月 12 日在 Open Intelligence Summit 前夕发布 Agent Learning Services 与 Agent Data Services，真正值得看的是它的叙事结构：先有 Inference Engine 让 Agent「会思考」，再有 Managed Agents 让它「能行动」，现在加上 Learning 与 Data Services 让它「会进步」。这份发布里最值得引用的是它自己对问题的描述——Agent 一旦上线就会进入平台期，而盯着它的那些工具只能告诉你系统还活着，答不上来「为什么它调错了工具」。下面讲清发布了什么、哪些还在预览后面，以及这条回路里不管买不买都该由你掌握的四个部分。

「一次比一次好」是这次发布真正想卖的东西

发布了什么

Agent Learning Services 由三个产品组成，合成一条回路：Insights 记录 Agent 做了什么；Signals 标出它在哪些地方做得不够；Evaluations & Simulations 让改动先跑一遍再给客户看。这三者产出的「证据」，就是团队用来优化 Agent 的依据，而不是靠猜着改提示词。Agent Data Services 是底座：一个托管的地方，让 Agent 读取它需要的上下文，并把学到的东西写回去。首席产品与技术官 Vinay Kumar 把这次变化概括为：问题已经从「能不能上线一个 Agent」，变成了「这个 Agent 会不会一次比一次好」。这组服务的定位是跑在客户自己的数据上、跑在开放的云上，而不是绑在某一家模型厂商的平台里。

# Insights, at its core, is a trace you own. Write it with redaction, not after.
import json, re, hashlib, time

SECRET = re.compile(r"(sk-[A-Za-z0-9]{8,}|Bearer\s+[A-Za-z0-9._-]+|\b\d{13,19}\b)")
FIELDS = {"email", "phone", "card", "address", "token", "api_key"}

def redact(value, path=""):
    if isinstance(value, dict):
        return {k: redact(v, k) for k, v in value.items()}
    if isinstance(value, list):
        return [redact(v, path) for v in value]
    if isinstance(value, str):
        if path in FIELDS:
            return "hash:" + hashlib.sha256(value.encode()).hexdigest()[:16]
        return SECRET.sub("[redacted]", value)
    return value

def trace(session_id, step, tool, args, result, tokens=None):
    return {
        "ts": round(time.time(), 3),
        "session_id": session_id,
        "step": step,
        "tool": tool,
        "args": redact(args, tool),
        "result_digest": hashlib.sha256(json.dumps(result, sort_keys=True).encode()).hexdigest()[:12],
        "result_chars": len(json.dumps(result)),
        "tokens": tokens,
    }

print(json.dumps(trace("s-1", 1, "search", {"q": "renewal date", "api_key": "sk-abcdef12345678"},
                       {"hits": 3}), indent=2))

# DigitalOcean describes this as assembling logs and agent sessions with
# configurable redaction controls. Configurable means decided before you write.

这次终于把「平台期」讲准了

这份发布自己对问题的描述异常诚实：Agent 一上线就进平台期，因为这项工作的每一块都散在别处。Agent 做过什么的记录躺在监控工具里，而那类工具显示的是「系统是否在线」，不是「它为什么调错了工具、为什么丢掉了用户」。修法通常就是改一改提示词，而改完没法测，于是第一个发现回归的人就是客户。想改进模型，就得把生产数据导出给训练厂商，再把四五个工具串起来，结果一个季度能做一次就算好的，多数时候根本不做。2026 年几乎所有把 Agent 放进生产的团队，都能在这四句话里认出两三条——这就是这套叙事成立的原因。

# Signals: flag the sessions that matter. Two cheap detectors catch most of them.
import json
from collections import Counter

def load(path):
    return [json.loads(line) for line in open(path) if line.strip()]

def looped(session, repeat_threshold=3, window=6):
    keys = [(s["tool"], json.dumps(s["args"], sort_keys=True)) for s in session]
    recent = keys[-window:]
    for key, count in Counter(recent).items():
        if count >= repeat_threshold:
            return {"kind": "loop", "tool": key[0], "repeats": count}
    return None

def stalled(session, token_budget=40000):
    spent = sum(s.get("tokens") or 0 for s in session)
    if spent > token_budget:
        return {"kind": "budget", "tokens": spent}
    if session and session[-1]["tool"] == "ask_user":
        return {"kind": "lost_user", "step": session[-1]["step"]}
    return None

sessions = {}
for row in load("traces.jsonl"):
    sessions.setdefault(row["session_id"], []).append(row)

flagged = {sid: (looped(s) or stalled(s)) for sid, s in sessions.items()}
for sid, signal in flagged.items():
    if signal:
        print(sid, signal)

# Each flagged session becomes a candidate case for Evaluations and Simulations.

把会话、信号、评估串成一条回路

Insights、Signals、Evaluations：回路本身

对于手工搭过评测框架的人来说，这条回路并不陌生，只是这次把顺序写明了。Insights 记录 Agent 的每一步。Signals 标出真正重要的会话——它在哪卡住、在哪打转、在哪丢掉用户、在哪失败。这些证据变成一次待验证的改动，Evaluations & Simulations 用那些被标出的会话构建出的真实对话，在真实用户之前先跑一遍。每一次改动再喂给回路的下一环。DigitalOcean 表示，针对模型的「递归式自我改进」今天就已经可用：增强后的 Inference Router 为每个任务挑选得分最高的模型，按成本、质量或团队自己的评测来优化，并在新模型发布时重跑那些评测。把这套自动回路扩展到 Agent 工作流被描述为下一步计划，形式是「增强」而不是取代人工审批。

# Evaluations: run a candidate change against the cases you harvested.
import json

def runner(prompt_version, case):
    # Stand-in for your agent. Swap in a real invocation of the candidate.
    expected = case["expect"]["tool"]
    known_good = {"v1": expected, "v2": expected if case["difficulty"] < 3 else "ask_user"}
    return {"tool": known_good.get(prompt_version, "ask_user"), "tokens": 1200}

def evaluate(cases, versions):
    table = {}
    for version in versions:
        passed = 0
        for case in cases:
            out = runner(version, case)
            if out["tool"] == case["expect"]["tool"]:
                passed += 1
        table[version] = round(100 * passed / len(cases), 1)
    return table

CASES = [
    {"id": "c1", "difficulty": 1, "expect": {"tool": "search"}},
    {"id": "c2", "difficulty": 2, "expect": {"tool": "update_ticket"}},
    {"id": "c3", "difficulty": 4, "expect": {"tool": "escalate"}},
]
print(evaluate(CASES, ["v1", "v2"]))

# The value is not the score. It is that the score is computed before a customer
# discovers the regression, from sessions that were already flagged in production.

底座那层数据服务

Agent Data Services 决定的是这条回路能不能跑起来，因为一次评测的质量上限，就是你能否重放当时的上下文。按 DigitalOcean 的描述，公司里的每一份数据在到达时就变成「Agent 可用」：视频被转写并生成描述，文档被读取并建索引，日志与 Agent 会话被组装起来，并带有可配置的敏感信息脱敏控制，而这一切都能在一次调用里按语义检索。数据落进一个逻辑上按客户隔离的目录，治理设计的目标是让每一份数据只被被授权的人与 Agent 触达。Signals 会把会话写进去，Optimizer 只在客户明确选择加入（opt in）时才从里面取训练与测试数据。同一层也被开放给普通的检索与分析用途，覆盖视频、音频、文档与结构化数据。

# Optimizer data is opt-in per customer. Encode the gate, do not document it.
import json
from datetime import date

CUSTOMERS = [
    {"tenant": "acme", "training_opt_in": True, "signed": "2026-09-01"},
    {"tenant": "northwind", "training_opt_in": False, "signed": None},
]

OPT_IN_VERSION = "2026-10"

def exportable(record, customer):
    if not customer["training_opt_in"]:
        return False, "no training opt-in on file"
    if record.get("contains_third_party_data"):
        return False, "session contains third-party data"
    if record["tenant"] != customer["tenant"]:
        return False, "tenant mismatch"
    return True, "ok"

def build_dataset(records, customers):
    index = {c["tenant"]: c for c in customers}
    dataset, skipped = [], []
    for rec in records:
        ok, why = exportable(rec, index[rec["tenant"]])
        (dataset if ok else skipped).append((rec["id"], why))
    return dataset, skipped

print(build_dataset([{"id": "r1", "tenant": "acme", "contains_third_party_data": False},
                     {"id": "r2", "tenant": "northwind", "contains_third_party_data": False}],
                    CUSTOMERS))

# "Only where the customer expressly opts in" is a query condition, not a policy page.

脱敏、留痕、复用，都在写入那一刻决定

哪些可用，哪些还不可用

Agent Learning Services（即 Insights、Signals 与 Evaluations & Simulations）目前处于公开预览。Inference Router 优化器与 Agent Data Services 处于私有预览，需要单独填表申请。发布中没有给出任何一方的公开定价，也没有独立验证过「这条回路确实让生产环境里的 Agent 变好」的基准数据。被点名的客户只有一家：Amplitude，其 CEO Spenser Skates 的说法是，他们按「实际交付效果」而不只是成本来为每个任务挑模型，并由 Managed Agents 承担扩容时的基础设施。可用的部分就按这个分界来看：可观测与评测今天就能试，数据层和路由优化器还不行。

# Tenant isolation: every read path should be filtered, including the debug path.
import json

RECORDS = [
    {"id": "r1", "tenant": "acme", "kind": "session"},
    {"id": "r2", "tenant": "acme", "kind": "document"},
    {"id": "r3", "tenant": "northwind", "kind": "session"},
]

ALLOWED_KINDS = {"acme": {"session", "document"}, "northwind": {"session"}}

def search(records, tenant, query="", kinds=None, actor="agent"):
    kinds = kinds or ALLOWED_KINDS[tenant]
    if kinds - ALLOWED_KINDS[tenant]:
        raise PermissionError("actor " + actor + " requested kinds outside the tenant grant")
    return [r for r in records
            if r["tenant"] == tenant and r["kind"] in kinds and query.lower() in json.dumps(r).lower()]

print(search(RECORDS, "acme", query=""))
print(search(RECORDS, "northwind", query=""))

# The same function should back your console, your API and your agent tool calls.
# An unfiltered path is a data leak with a friendly interface.

不管买不买，都该自己掌握的四个部分

不管你是否采购这些服务，这条回路都由四个部件构成，而它们的接口你本来就有。第一，在调用时刻就写下的持久轨迹，并且把脱敏放在写入口而不是下游任务里——因为你没脱掉的那份数据，就是你再也撤不回来的那份。第二，一个针对两种廉价失败的信号检测器：打转和超预算；这两类用计数器就能发现，不需要模型。第三，一个可重放的评测集，素材来自被标记的会话，这样一次提示词改动在「有用户」之前就已经有了分数。第四，在任何把生产会话变成训练或测试数据的路径上，加一道显式的选择加入校验，并且用代码强制，而不是写在政策页上。