FDE 交付实战：把模型塞进客户机房的那套工程纪律

原创

资源大佬 jzit-top

发布于 2026-10-23 14:46:36

520

Forward Deployed Engineer（FDE）的交付，本质不是"部署一个产品"，而是在客户真实的、脏的、异构的环境里，造出一个能跑、能评、能回滚的系统。产品团队的代码跑在干净的云上，FDE 的代码要跑在客户的 Oracle 数据库、内网 FTP 和没开外网的 Kubernetes 里。本文从交付链路拆解 FDE 的核心工程动作，并给出可复用的代码骨架。

一、交付链路的四个阶段

一条典型的 FDE 交付链是：数据接入 → 业务编排 → 离线评测 → 上线监控。四段中任何一段缺失，交付都会在客户验收时崩掉。

1. 数据接入：适配器模式吃掉异构性

客户数据源永远是异构的。FDE 的第一条纪律是：所有数据源统一收敛到一个窄接口，业务逻辑绝不直接碰客户数据库。

from abc import ABC, abstractmethod
from dataclasses import dataclass
import pandas as pd

@dataclass
class Record:
    id: str
    text: str
    meta: dict

class SourceAdapter(ABC):
    @abstractmethod
    def fetch(self, since: str, limit: int) -> list[Record]: ...

class SqlAdapter(SourceAdapter):
    def __init__(self, conn, table: str):
        self.conn, self.table = conn, table
    def fetch(self, since: str, limit: int) -> list[Record]:
        df = pd.read_sql(
            f"SELECT id, body AS text, created_at FROM {self.table} "
            f"WHERE created_at >= %(since)s ORDER BY created_at LIMIT %(limit)s",
            self.conn, params={"since": since, "limit": limit})
        return [Record(r.id, r.text, {"ts": str(r.created_at)}) for r in df.itertuples()]

class FtpCsvAdapter(SourceAdapter):
    def __init__(self, path: str): self.path = path
    def fetch(self, since: str, limit: int) -> list[Record]:
        df = pd.read_csv(self.path).query("created_at >= @since").head(limit)
        return [Record(str(r.id), r.body, {"ts": r.created_at}) for r in df.itertuples()]

要点：since + limit 的增量拉取签名让重跑变得幂等；Record 是唯一跨模块契约，换数据源不改下游。

2. 业务编排：结构化输出必须校验

客户验收看的不是"模型说了什么"，而是"字段填对没有"。LLM 输出必须过 schema 校验，失败重试而非静默放行。

import json
from pydantic import BaseModel, ValidationError

class Ticket(BaseModel):
    category: str
    urgency: int
    summary: str

def classify(llm, text: str, retries: int = 2) -> Ticket:
    for i in range(retries + 1):
        raw = llm.complete(
            system="只输出JSON，字段: category, urgency(1-5), summary",
            user=text, temperature=0)
        try:
            return Ticket(**json.loads(raw))
        except (json.JSONDecodeError, ValidationError) as e:
            if i == retries:
                raise RuntimeError(f"schema failed: {e}") from e
    raise AssertionError("unreachable")

要点：temperature=0 保证可复现；校验失败抛异常让上游进入死信队列，而不是把脏数据写进客户系统——这是 FDE 和"demo 工程师"的分水岭。

3. 离线评测：交付前必须有数字

没有评测集的交付等于赌博。FDE 到现场第一周就要和客户业务方一起标注 100–300 条黄金样本，固化成回归测试。

def evaluate(pipeline, gold: list[dict]) -> dict:
    correct_cat = correct_urg = 0
    for g in gold:
        pred = pipeline(g["text"])
        correct_cat += pred.category == g["category"]
        correct_urg += abs(pred.urgency - g["urgency"]) <= 1
    n = len(gold)
    return {"category_acc": correct_cat / n, "urgency_within1": correct_urg / n}

要点：评测指标要能被业务方读懂（"分类准确率"而非"logloss"）；指标与验收条款直接挂钩，写进交付物清单。

4. 上线监控：可回放是底线

线上出现 badcase 时，FDE 要能在本地用同样的输入复现。所有请求必须落盘 input + prompt_version + model + output。

import hashlib, time, json

def traced_run(pipeline, text: str, log_path: str):
    out = pipeline(text)
    rec = {
        "ts": time.time(),
        "input_hash": hashlib.sha256(text.encode()).hexdigest()[:16],
        "prompt_version": pipeline.version,
        "model": pipeline.model_name,
        "output": out.model_dump(),
    }
    with open(log_path, "a") as f:
        f.write(json.dumps(rec, ensure_ascii=False) + "\n")
    return out

二、FDE 的交付物清单

代码之外，FDE 交付的硬性产物包括：数据契约文档、黄金评测集、回滚 Runbook、监控看板、以及一份"已知失败模式"清单。最后一项最容易被忽略，却是客户信任的来源——诚实列出模型会在哪些边界场景出错，比宣称 99% 准确率更能促成验收。

FDE 的工程哲学可以概括为：用产品团队的工程标准，解决客户现场的非标问题。适配器收敛异构性、schema 校验守住数据质量、评测集锚定验收标准、可回放日志支撑线上排障——这四条纪律，比任何模型选型都更决定交付成败。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

项目实战