本地优先Agent的四个信号决策门

一位后端工程师登上了末班列车，笔记本电脑的终端里还跑着一个未完成的 agent 任务。机器的本地保险库里存着各种仓库 token，而下一个工具调用需要一次漫长的编译，会让风扇全力运转。蜂窝数据在两格信号和飞行模式之间来回切换，所以纯远程的方案随时会毫无预警地卡住。纯本地的方案则会让笔记本在下一站到达前就热得发烫、电量耗尽。

这段通勤行程正是仍在推行本地优先 agent 的团队所面临的完整缩影。有些工作永远不能离开磁盘，因为密钥和客户数据片段就放在 checkout 旁边。有些工作没有额外算力根本完成不了，因为主机已经发热、电池受限或处于温控降频状态。网络不是承诺，延迟是一个经过计量的往返时间——要么比本地加载更快，要么就不行。

Placement 属于控制平面层面的决策，而不是 prompt 前言或供应商复选框能搞定的事。本文其余部分将把四个信号当作可执行策略来处理，然后给出决策表、一个轻量的 Python gate 以及锁定策略的测试。读者可以删除所有产品名称，仍然保留一套可用的本地与远程切分方案。

Always-local 和 Always-remote 都会失败

Always-local 看似安全，直到冷启动权重把第一个有效 token 的出现时间拖到用户注意力早已转移之后。Always-remote 看似可扩展，直到 API 密钥、私有补丁或断开的隧道把 agent 变成一块砖。每次出事故、出差、笔记本休眠，使用单一端点包裹所有工具调用的团队都会领教到这一点。真正有用的切分是：在声明的延迟预算下，哪些通道可以承载哪些负载，同时密钥必须留在本机。

Placement gate 应该刻意做得无聊。它读取四个信号、返回一个通道、在这四个信号冲突时拒绝即兴发挥。民间流传的路由器如果在一条 if 语句里混合了任务类型、队列深度和模型品牌，就会让密钥泄漏进重试缓冲区。

四个真正能改变通道的信号

第一个信号是延迟，以一次短探测来衡量，而不是拿上周的亲身经历当依据。本地加载时间包括磁盘、内存压力和模型在温热主机上的启动；远程时间包括 DNS、TLS 和一次廉价健康检查请求。如果探测失败，说明远程通道不存在——这和「只是慢」是完全不同的状态。远程不可用时必须 fail closed，而不是 fail 进排队的上传里。

第二个信号是密钥级别，比「用户曾经输入过一次密码」这种复选框要严格得多。工作信封如果仍然包含 token、私钥、客户记录或未发布的补丁，就必须驻留在磁盘上。已经剥离到只剩公开 issue 编号、文件哈希或经编辑的堆栈跟踪的信封可以传输。未知级别按保险库级别处理，因为一个猜出来的标签会让私有代码搭上突发通道。

第三个信号是离线能力，意思是任务有一个本地降级方案，仍能取得可衡量的进展。lint 流程、单元测试或基于磁盘索引的检索可以在广域网在站点之间断开时继续。没有本地降级的长时间合成任务不在此列，gate 应该如实说明，而不是假装有。远程才能存在的进展不是离线能力。

第四个信号是突发需求，考察的是主机容量而不是偏好用别人的 GPU。突发意味着电池低于本地策略阈值、机身受热限制、或工作量超过了声明的本地分钟上限。突发是免费远程服务器唯一被允许胜出的时刻，而且仅针对已经通过密钥检查的信封。其他所有场景下，本地优先仍是默认值，因为数据从未移动过。

当这四个信号不一致时，策略刻意保持保守。密钥即使在突发请求强烈且探测看起来完美的情况下也会把任务钉在本机。没有本地降级的离线任务会 fail closed，而不是把负载复制进崩溃日志。无法测量的延迟按远程不可用处理，而不是按远程更快处理。

运行时可执行的决策表

这张表是代码必须遵守的产物。行按顺序评估，第一个匹配者胜出，这保持了实现免于嵌套民间传说。

local_preferred 仍在笔记本上运行，因为主机可以在自身预算内完成工作。remote_burst 是唯一可以消耗远程运行时的通道，而且永远不会收到原始保险库。fail_closed 是一个特性：它胜过那种把密钥序列进调试转储的静默重试。团队应该记录通道名称和探测原因，绝不记录负载正文。

实现：一个轻量的 placement gate

以下模块旨在保存在 agent wrapper 旁边并在本地执行。它不调用任何供应商，远程执行仍是一个桩，以便读者后续交换自己的传输层。把探测当作健康检查来用，而不是跑分研究。

# placement_gate.py
from __future__ import annotations

from dataclasses import dataclass
from enum import Enum
from typing import Optional
import json
import time
import urllib.error
import urllib.request

class Lane(str, Enum):
    LOCAL_ONLY = "local_only"
    LOCAL_FALLBACK = "local_fallback"
    LOCAL_PREFERRED = "local_preferred"
    REMOTE_BURST = "remote_burst"
    FAIL_CLOSED = "fail_closed"

@dataclass(frozen=True)
class JobEnvelope:
    job_id: str
    secret_class: str  # vault | private_source | sanitized
    has_local_fallback: bool
    burst_needed: bool
    payload: str

@dataclass(frozen=True)
class ProbeResult:
    ok: bool
    rtt_ms: Optional[float]
    reason: str

BANNED_FRAGMENTS = (
    "begin private",
    "api_key",
    "authorization:",
    "x-api-key",
    "secret=",
)

def classify_secret(envelope: JobEnvelope) -> bool:
    """Return True only when the envelope is allowed to leave the disk."""
    if envelope.secret_class in {"vault", "private_source"}:
        return False
    if envelope.secret_class != "sanitized":
        return False
    lowered = envelope.payload.lower()
    return not any(fragment in lowered for fragment in BANNED_FRAGMENTS)

def redact_for_wire(envelope: JobEnvelope) -> JobEnvelope:
    if not classify_secret(envelope):
        raise ValueError("refusing to copy a non-sanitized envelope onto the wire")
    # Keep public identifiers; drop anything that looks like assignment of secrets.
    kept_lines = []
    for line in envelope.payload.splitlines():
        lower = line.lower()
        if any(fragment in lower for fragment in BANNED_FRAGMENTS):
            continue
        kept_lines.append(line)
    return JobEnvelope(
        job_id=envelope.job_id,
        secret_class="sanitized",
        has_local_fallback=envelope.has_local_fallback,
        burst_needed=envelope.burst_needed,
        payload="\n".join(kept_lines).strip(),
    )

def probe_remote(url: str, timeout_s: float = 1.5) -> ProbeResult:
    started = time.perf_counter()
    try:
        req = urllib.request.Request(url, method="HEAD")
        with urllib.request.urlopen(req, timeout=timeout_s) as resp:
            _ = resp.status
        rtt = (time.perf_counter() - started) * 1000.0
        return ProbeResult(ok=True, rtt_ms=rtt, reason="probe_ok")
    except (urllib.error.URLError, TimeoutError, ValueError) as exc:
        return ProbeResult(ok=False, rtt_ms=None, reason=type(exc).__name__)

def place(envelope: JobEnvelope, probe: ProbeResult) -> Lane:
    if not classify_secret(envelope):
        return Lane.LOCAL_ONLY
    if not probe.ok and envelope.has_local_fallback:
        return Lane.LOCAL_FALLBACK
    if not probe.ok:
        return Lane.FAIL_CLOSED
    if envelope.burst_needed:
        return Lane.REMOTE_BURST
    return Lane.LOCAL_PREFERRED

def decide(envelope: JobEnvelope, probe_url: str) -> dict:
    probe = probe_remote(probe_url)
    lane = place(envelope, probe)
    wire = None
    if lane is Lane.REMOTE_BURST:
        wire = redact_for_wire(envelope).payload
    return {
        "job_id": envelope.job_id,
        "lane": lane.value,
        "probe_ok": probe.ok,
        "probe_reason": probe.reason,
        "wire_payload": wire,
    }

if __name__ == "__main__":
    demo = JobEnvelope(
        job_id="job-4412",
        secret_class="sanitized",
        has_local_fallback=True,
        burst_needed=True,
        payload="public issue 4412: flaky timer in worker",
    )
    print(json.dumps(decide(demo, "https://example.com"), indent=2))

在通勤网络环境下，一次简单的 HEAD 请求足以区分「远程存在」和「远程是幻想」。它不足以宣称某个运行时比另一个更快，本文也不会发布谁胜谁负、限额或硬件对比。把 probe_url 指向团队实际运维的健康端点，而不是某个随机的第三方页面。

采用 gate 的编号工作流

在 agent store 中清点密钥级别，在任何模型调用之前把每个工具映射到 vault、private_source 或 sanitized。

在一个纯函数中剥离远程信封，在 DNS 被触及之前就丢弃头部、token 和未发布的代码块。

添加一个属于团队的探测 URL，当在列车网络环境下 DNS、TLS 或超时触发时 fail closed。

从电池百分比、热状态或主机已知的一个简单已耗分钟上限来声明本地突发策略。

只把经过 sanitized 的突发任务路由到远程运行时，把原始信封保留在磁盘上以便后续本地恢复。

记录通道决策但不记录负载正文，然后在启用流量之前为五张表行添加测试。

第五步是免费远程选项在这个设计中唯一应该出现的地方。当笔记本无法吸收突发工作时，拥有免费模型访问和免费服务器选项的远程运行时可以接收已经编辑过的任务。披露：本文是 MonkeyCode 产品推广的一部分。Gate 不依赖该运行时；它只需要一个接受已编辑信封并忽略 local_only 中所有内容的传输层。

锁定策略的测试

# test_placement_gate.py
from placement_gate import (
    JobEnvelope,
    Lane,
    ProbeResult,
    classify_secret,
    place,
    redact_for_wire,
)

DOWN = ProbeResult(ok=False, rtt_ms=None, reason="timeout")
UP = ProbeResult(ok=True, rtt_ms=84.0, reason="probe_ok")  # fixture, not a measurement claim

def env(**kwargs) -> JobEnvelope:
    base = dict(
        job_id="job-1",
        secret_class="sanitized",
        has_local_fallback=True,
        burst_needed=False,
        payload="public issue 4412: flaky timer in worker",
    )
    base.update(kwargs)
    return JobEnvelope(**base)

def test_vault_never_leaves_even_when_burst_and_probe_are_ready():
    job = env(secret_class="vault", burst_needed=True, payload="token=not-for-wire")
    assert place(job, UP) is Lane.LOCAL_ONLY

def test_private_source_stays_local_when_offline():
    job = env(secret_class="private_source", has_local_fallback=False)
    assert place(job, DOWN) is Lane.LOCAL_ONLY

def test_offline_sanitized_job_uses_local_fallback():
    job = env(has_local_fallback=True)
    assert place(job, DOWN) is Lane.LOCAL_FALLBACK

def test_offline_without_fallback_fails_closed():
    job = env(has_local_fallback=False)
    assert place(job, DOWN) is Lane.FAIL_CLOSED

def test_no_burst_stays_local_even_if_remote_is_healthy():
    job = env(burst_needed=False)
    assert place(job, UP) is Lane.LOCAL_PREFERRED

def test_sanitized_burst_may_use_remote():
    job = env(burst_needed=True)
    assert place(job, UP) is Lane.REMOTE_BURST

def test_payload_with_key_material_is_not_sanitized():
    job = env(payload="Authorization: Bearer demo")
    assert classify_secret(job) is False
    assert place(job, UP) is Lane.LOCAL_ONLY

def test_redact_refuses_vault_envelopes():
    job = env(secret_class="vault", payload="not-for-wire")
    try:
        redact_for_wire(job)
    except ValueError:
        return
    raise AssertionError("vault envelopes must not be copied for the wire")

用一套无聊的工具链来运行契约，这样在下次重构之后策略仍然可执行：

python -m pip install pytest
python -m pytest test_placement_gate.py -q
python placement_gate.py

测试即契约。如果未来的 wrapper 为模型品牌或队列深度增加了额外状态，这些行仍应在任何人争论 token 计费方式之前通过。84.0 毫秒是健康探测的 fixture，不是记录的通勤测量值，也不是产品声明。

局限性与适用人群

Gate 不能让远程供应商对受监管数据变得安全。如果合同禁止甚至编辑过的堆栈跟踪离开大楼，就必须把 remote_burst 编译掉，而不是仅在运行时跳过。没有本地密钥存储的团队根本不应该启用 remote_burst，因为他们无法证明信封是经过 sanitized 的。整个工作负载只是在冷却机器上跑短 lint 循环的团队应该留在 local_preferred 并跳过这些额外的移动部件。需要保证离线完成的团队必须先构建降级方案；gate 不会发明本地编译器、本地索引或本地模型启动路径。

免费服务器仅在剩余场景中胜出：sanitized 负载、健康探测、以及主机诚实地无法完成工作。在那个切片中延迟仍然重要，但它是作为存在性检查，而不是排行榜。密钥仍在每次冲突中获胜——这正是保持驻留本地、只溢出算力的全部意义。

Placement 是控制平面的决策，而不是 agent 循环下面的模型的人格特质。列车故事的结局和大多数事故一样：工程师要么把保险库留在磁盘上并完成了本地降级，要么等待探测并溢出了一个已编辑的突发任务。把模块丢进 agent wrapper、运行 pytest 文件、拒绝表中未列出的任何通道。

已经将密钥固定到本地磁盘的开发者可以把经过 sanitized 的 remote_burst 通道附加到那个免费服务器选项上，并在相同的公开任务上将结果与 local_preferred 对比。