在这里插入图片描述

之前小马也写过一篇《FDE 到底需要什么技术栈?从全栈工程到 AI 落地的复合能力》,简单介绍了FDE前沿部署工程师到底需要具备哪些技能,前段时间也学习了腾讯云的FDE整个系列的教程(但是没交钱考证哈^^),今天我们聊聊更细节的内容。
FDE(Forward Deployed Engineer,前线部署工程师)最早由 Palantir 提出,工程师直接驻场或贴身对接客户,把产品落地到客户的真实业务里。近两年,AI 公司(大模型厂商、Agent 平台、各类 AI 应用公司)也在大量招这个岗位。

本文整理两个问题:

  1. FDE 到底需要什么技能栈?
  2. 如果你是 FDE,应该怎么一步步实现一个系统?

说明:文中的技能栈和流程是按行业通行的岗位描述和实践整理的,不针对某一家公司,具体要求以实际 JD 为准。


一、FDE 本质上做什么

一句话:既要写代码,又要面对客户,把通用产品改造成能解决客户具体问题的方案。

典型工作流:

  1. 下沉到客户现场,理解业务和痛点
  2. 梳理客户的数据、系统和流程
  3. 快速做出原型或 POC,并上线交付
  4. 把客户反馈带回产品团队,推动产品改进
  5. 持续运维、培训,扩大合作

二、技能栈

2.1 工程能力(硬门槛)

方向具体内容
编程语言Python 为主,TypeScript / JavaScript 常用,Java / Go 视公司而定
后端与 APIREST / GraphQL、鉴权、异步任务、微服务基础
数据工程SQL、ETL / 数据管道、数据建模、Spark 等大数据组件
前端基础能快速做出可演示的界面(React 等),不要求精通
云与部署Docker、Kubernetes、AWS / Azure / GCP 或国内云、CI/CD
系统集成对接客户的 ERP、CRM、数据库、内部系统,处理脏数据和老旧接口

特点是要广不要极深,强调落地速度。

2.2 AI / Agent 能力(AI 方向 FDE 的核心)

  • 大模型应用:Prompt 工程、函数调用、结构化输出
  • RAG:切片、Embedding、混合检索、重排
  • Agent:规划、工具调用、记忆、多 Agent 编排,LangChain / LlamaIndex / 自研框架
  • 评测与调优:评测集构建、LLM-as-Judge、任务成功率、回归测试
  • 模型选型、成本与延迟优化、微调基础
  • 安全与合规:权限隔离、Prompt 注入、数据脱敏

2.3 业务与客户能力(区别于普通工程师的关键)

能力说明
需求挖掘听出客户「说的」和「真正要的」之间的差别
业务理解快速进入金融、制造、政务、医疗、游戏等垂直领域
问题拆解把模糊的业务问题拆成可实现的技术方案
沟通表达向业务方和高管讲清方案,也能和研发讲清需求
项目管理控范围、管预期、拉齐多方,按时交付
售前与 POC配合销售做方案、演示、验证价值

2.4 软性素质

  • 抗压和自驱:常常一个人面对复杂现场
  • 结果导向:关注客户价值,而不只是技术漂亮
  • 快速学习:每个客户的行业、系统都不一样
  • 出差与驻场的适应力
  • Ownership:从方案到上线到验收,全程负责

三、与相近岗位的区别

岗位侧重
普通后端 / 平台工程师做通用产品,不直接面对客户
解决方案架构师(SA)偏方案设计和售前,写代码较少
售前工程师偏演示和投标
实施 / 交付工程师偏配置和部署,开发深度较浅
FDE既要写代码做定制开发,又要深度参与客户业务

四、不同阶段怎么准备

入门:Python + SQL + 一个完整的小项目,如 RAG 问答或带工具调用的 Agent,部署上线并写评测。

进阶:

  • 做一个能对接真实业务系统的项目(数据库、第三方 API、权限)
  • 补齐云部署和监控
  • 练习把项目讲成「业务价值 + 技术方案」,而不只是技术细节

加分项:

  • 有咨询、售前、产品或创业经历
  • 熟悉某个垂直行业
  • 有对外交付或开源作品
  • 英文沟通能力(外企或出海岗位)

五、判断自己是否适合

  • 适合:喜欢把技术用到真实场景,享受解决客户问题的成就感,不排斥沟通和出差。
  • 不太适合:只想专注写底层代码,或者不喜欢不确定性和频繁切换场景。

六、FDE 实现一个系统的完整过程

核心特点:小步快跑、现场迭代、先证明价值再扩大,而不是传统的「需求评审、开发、测试、上线」瀑布流。

6.1 整体流程

进场调研 → 定义问题与成功标准 → 数据与系统摸底 → 快速原型(POC)
→ 评测与迭代 → 生产化上线 → 培训与推广 → 反馈回产品

6.2 阶段 1:进场调研(1 到 2 周)

目标:搞清楚客户真正的问题,而不是客户说的需求。

  • 和一线业务人员(不只是管理者)聊,亲眼看他们现在怎么干活
  • 梳理现有流程:谁在做、花多久、哪里最痛、出错代价多大
  • 识别干系人:出钱的人、使用的人、拍板的人、会阻碍的人(安全、IT、法务)
  • 产出:业务流程图、痛点清单、候选场景列表

常见误区:客户说「要一个智能客服」,真实痛点可能是「工单分派慢」。

6.3 阶段 2:定义问题与成功标准

这一步最关键,决定后面所有评测。

  • 选一个小而有价值、可衡量的场景起步,不要一上来做大而全
  • 把业务目标变成可量化指标,例如:
    • 处理时长从 30 分钟降到 5 分钟
    • 任务成功率不低于 85%
    • 人工介入率下降 50%
  • 约定边界:哪些做,哪些不做,哪些必须有人审核
  • 产出:一页纸的目标、指标、范围和风险

建议把这一页纸写成机器可读的配置,后面的评测脚本直接读取它,避免「口头约定」:

scenario: 需求评审辅助
goal: 评审前自动生成初审报告
metrics:
  - name: task_success_rate        # 任务成功率
    target: ">= 0.85"
  - name: avg_review_minutes       # 单次评审耗时(分钟)
    baseline: 30
    target: "<= 18"
  - name: human_override_rate      # 人工改写比例
    target: "<= 0.3"
scope:
  in:  [读取需求, 对照规范, 输出评审意见]
  out: [自动修改需求, 自动流转状态]       # 明确不做的事
human_review_required:
  - 写入评审结论                        # 关键操作必须有人确认

6.4 阶段 3:数据与系统摸底

多数项目的真实工作量在这里,不在模型。

摸底项要问的问题
数据在哪、什么格式、质量如何、能不能用、有无敏感信息
系统有哪些接口、有没有 API、鉴权方式、调用限制
权限不同角色能看什么,Agent 能操作什么
合规数据能不能出域、能不能用外部模型、是否需要脱敏
环境部署在云上还是内网,网络是否隔离

产出:数据清单、接口清单、权限与合规约束。必要时做数据清洗和接入管道。

摸底阶段建议写一个一键体检脚本,每到一个新客户就跑一遍,快速暴露连通性、延迟、数据质量和敏感信息问题:

import re, time, requests

SENSITIVE_PATTERNS = {
    "手机号": re.compile(r"1[3-9]\d{9}"),
    "身份证": re.compile(r"\d{17}[\dXx]"),
    "邮箱":   re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"),
}

def check_endpoint(name, url, headers=None, timeout=5):
    """检查接口:能否连通、鉴权是否通过、延迟多少"""
    start = time.time()
    try:
        r = requests.get(url, headers=headers, timeout=timeout)
        return {"name": name, "ok": r.ok, "status": r.status_code,
                "latency_ms": int((time.time() - start) * 1000)}
    except Exception as e:
        return {"name": name, "ok": False, "error": str(e)}

def profile_records(records, required_fields):
    """数据质量:字段缺失率 + 敏感信息命中数"""
    total = len(records) or 1
    missing = {f: sum(1 for r in records if not r.get(f)) / total
               for f in required_fields}
    hits = {k: 0 for k in SENSITIVE_PATTERNS}
    for r in records:
        text = " ".join(str(v) for v in r.values())
        for k, p in SENSITIVE_PATTERNS.items():
            if p.search(text):
                hits[k] += 1
    return {"count": len(records), "missing_rate": missing, "sensitive_hits": hits}

if __name__ == "__main__":
    print(check_endpoint("需求接口", "https://example.com/api/requirements/1"))
    sample = fetch_sample_records(limit=200)          # 换成客户系统的取数方式
    print(profile_records(sample, ["title", "body", "owner"]))

输出里的缺失率和敏感信息命中数,可以直接放进「数据清单」,也是和客户谈脱敏、权限方案的依据。

6.5 阶段 4:快速原型(POC,1 到 3 周)

目标:尽快让客户看到能跑的东西,验证价值,不求完美。

技术选型思路:

业务问题
  ├─ 需要查资料、问答?      → RAG
  ├─ 需要执行操作、多步骤?   → Agent + 工具调用
  ├─ 流程固定、步骤明确?     → 工作流编排(优先,更稳定)
  └─ 需要审核、有风险?       → 人在回路(Human-in-the-loop)

要点:

  • 能用工作流就不用全自主 Agent,越可控越容易上线
  • 先用最强模型把效果做出来,再考虑降本
  • 工具接口尽量封装得简单、清晰,减少 Agent 出错
  • 一开始就保留完整 Trace,后面评测和排障都靠它
  • 做可演示的界面,让业务方能亲手试

下面是一个工作流 + 工具封装 + Trace + 人在回路的最小骨架。流程由代码固定,只有「评审意见生成」这一步交给大模型,所以稳定、好评测:

import time

class Trace:
    """记录每一步调用,评测和排障都靠它"""
    def __init__(self):
        self.records = []

    def call(self, name, fn, *args, **kwargs):
        start, ok = time.time(), False
        try:
            out = fn(*args, **kwargs)
            ok = True
            return out
        finally:
            self.records.append({
                "name": name,
                "args": {"args": args, "kwargs": kwargs},
                "ok": ok,
                "ms": int((time.time() - start) * 1000),
            })

# ---- 工具:入参简单、返回结构固定,减少 Agent 出错 ----
def fetch_requirement(req_id: str) -> dict:
    data = http_get(f"/api/requirements/{req_id}")     # 换成客户系统的接口
    return {"id": req_id, "title": data["title"], "body": data["body"]}

def load_review_rules() -> str:
    return read_file("rules/review_checklist.md")

# ---- 工作流:步骤固定,只有一步用大模型 ----
def review_workflow(req_id: str):
    trace = Trace()
    req = trace.call("fetch_requirement", fetch_requirement, req_id)
    rules = trace.call("load_review_rules", load_review_rules)

    prompt = f"""你是需求评审助手。
评审规范:
{rules}

需求标题:{req['title']}
需求正文:
{req['body']}

请从完整性、可行性、风险点三个方面给出评审意见,并标出不确定之处。"""
    draft = trace.call("llm_review", call_llm, prompt)

    return {
        "draft": draft,
        "need_human_confirm": True,       # 人在回路:不直接写入,由评审人确认
        "trace": trace.records,
    }

如果任务确实需要模型自主决定调用哪个工具,再升级成 Agent;升级时保留同一个 Trace,评测代码不用改。

6.6 阶段 5:评测与迭代(贯穿始终)

  1. 从客户真实数据里挑 30 到 100 条典型任务,和业务专家一起标注期望结果
  2. 工具调用用规则检查点,回答质量用 LLM Judge,会改状态的任务做状态校验
  3. 跑出基线,分析失败样本,归类(检索失败、工具选错、幻觉、权限问题)
  4. 针对 Top 失败原因迭代:改 Prompt、调检索、改工具描述、加规则兜底
  5. 每次改动自动回归,防止修好一个坏了另一个
  6. 把评测结果直接给客户看,用数据说话

产出:评测集、基线报告、迭代记录。

评测用例可以用「检查点」来写:必需检查点做门禁,决定这次任务通不通过;可选检查点和回答检查点参与打分,反映质量。

{
  "id": "case_001",
  "input": "请评审需求 REQ-1001",
  "tool_checkpoints": [
    {"id": "t1", "tool": "fetch_requirement", "required": true,
     "reason": "必须先拉取需求正文,否则评审没有依据"},
    {"id": "t2", "tool": "load_review_rules", "required": true,
     "reason": "必须加载评审规范"},
    {"id": "t3", "tool": "read_attachment", "required": false,
     "reason": "需求可能带附件,非必需路径"}
  ],
  "answer_checkpoints": [
    {"id": "a1", "desc": "评审意见覆盖完整性、可行性、风险点", "weight": 2, "required": true},
    {"id": "a2", "desc": "对不确定之处明确标注,而不是编造", "weight": 1, "required": false}
  ]
}

配套的最小评分脚本:工具检查点用规则判,回答检查点交给 LLM Judge 逐条打分:

def score_tool(cp, trace):
    hit = any(r["name"] == cp["tool"] and r["ok"] for r in trace)
    return {"id": cp["id"], "hit": hit, "score": 100 if hit else 0}

JUDGE_PROMPT = """只评判下面这一个检查点。
回答:{answer}
检查点:{desc}
先写理由,再给 0-100 分,只输出 JSON:{{"reason": "...", "score": 0}}"""

def score_answer(cp, answer):
    res = call_judge(JUDGE_PROMPT.format(answer=answer, desc=cp["desc"]))
    return {"id": cp["id"], "score": res["score"], "weight": cp["weight"]}

def evaluate_case(case, out):
    tool_res = [score_tool(c, out["trace"]) for c in case["tool_checkpoints"]]
    ans_res = [score_answer(c, out["draft"]) for c in case["answer_checkpoints"]]

    # 门禁:必需的工具检查点必须命中;必需的回答检查点不低于 60 分
    passed = all(r["hit"] for r, c in zip(tool_res, case["tool_checkpoints"]) if c["required"]) \
         and all(r["score"] >= 60 for r, c in zip(ans_res, case["answer_checkpoints"]) if c["required"])

    wsum = sum(r["weight"] for r in ans_res)
    quality = sum(r["score"] * r["weight"] for r in ans_res) / wsum
    return {"passed": passed, "quality": quality}

# 同一用例跑多次,统计稳定性(pass^k:k 次全部通过才算稳定)
def pass_hat_k(case, k=3):
    return all(evaluate_case(case, review_workflow(case["req_id"]))["passed"] for _ in range(k))

每次改 Prompt、换模型、调检索后跑一遍,把「通过率、平均质量分、pass^k、耗时、Token」做成对比表,直接拿给客户看。

6.7 阶段 6:生产化上线

POC 能跑和能上线之间差距很大,要补齐:

方面内容
稳定性超时重试、降级、限流、多次运行一致性(pass^k)
安全权限隔离、Prompt 注入防护、敏感数据脱敏、操作审计
可观测日志、Trace、监控告警、成本统计
人工兜底低置信度转人工、关键操作二次确认
部署容器化、CI/CD、灰度发布、可回滚
成本Token 用量、缓存、模型分级使用

上线策略:先灰度、先影子模式(Agent 只给建议,人来决定),再逐步放权。

下面是几个最常用的生产化组件:重试与超时、权限校验、审计日志、影子模式和失败兜底。

import functools, json, logging, time

logger = logging.getLogger("agent")

# 1) 重试 + 指数退避:应对工具或模型的偶发失败
def retry(times=3, backoff=1.0, exceptions=(Exception,)):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(*args, **kwargs):
            for i in range(times):
                try:
                    return fn(*args, **kwargs)
                except exceptions:
                    if i == times - 1:
                        raise
                    time.sleep(backoff * (2 ** i))
        return wrapper
    return deco

# 2) 权限校验:Agent 能做什么,取决于当前用户能做什么,而不是 Agent 自己
def require_permission(action):
    def deco(fn):
        @functools.wraps(fn)
        def wrapper(user, *args, **kwargs):
            if action not in get_user_permissions(user):
                raise PermissionError(f"用户 {user} 无权执行 {action}")
            return fn(user, *args, **kwargs)
        return wrapper
    return deco

# 3) 审计日志:谁、什么时候、对什么、做了什么,必须可追溯
def audit(event, **fields):
    logger.info(json.dumps({"event": event, "ts": int(time.time()), **fields},
                           ensure_ascii=False))

# 4) 影子模式开关:先只给建议,不真正写入
SHADOW_MODE = True

@require_permission("review:write")
def write_review_result(user, req_id, result):
    if SHADOW_MODE:
        audit("shadow_write", user=user, req_id=req_id, preview=result[:200])
        return {"written": False, "suggestion_only": True}
    save_to_system(req_id, result)                     # 换成客户系统的写入方式
    audit("write", user=user, req_id=req_id)
    return {"written": True}

# 5) 兜底:大模型失败或置信度低时转人工,而不是硬答
@retry(times=3, backoff=1.0)
def safe_review(req_id):
    return review_workflow(req_id)

def handle_request(user, req_id):
    try:
        out = safe_review(req_id)
    except Exception as e:
        audit("fallback_to_human", user=user, req_id=req_id, error=str(e))
        return {"status": "转人工", "reason": "自动评审失败"}
    return {"status": "待人工确认", "draft": out["draft"]}

灰度放权的节奏可以用配置驱动:

rollout:
  stage_1: {mode: shadow,   users: ["pilot_group"], write: false}   # 只给建议
  stage_2: {mode: assisted, users: ["pilot_group"], write: true, need_confirm: true}
  stage_3: {mode: auto,     users: ["all"],         write: true, need_confirm: false,
            guard: "任务成功率 >= 0.9 且 人工改写率 <= 0.2"}

每一阶段都要有明确的放权条件,用第 6.6 节的评测数据来判断,而不是凭感觉。

6.8 阶段 7:培训与推广

  • 给一线用户做培训,收集使用习惯和抱怨
  • 写使用手册和常见问题
  • 找出「种子用户」带动其他人
  • 追踪实际使用数据:使用率、采纳率、节省的时间

技术上成功但没人用,等于失败。

6.9 阶段 8:反馈回产品

  • 把客户场景里共性的需求沉淀成产品功能或通用组件
  • 把一次性的定制代码抽象为可复用模块
  • 反馈产品缺陷和能力缺口

这也是 FDE 价值的放大点:一个客户的经验,变成所有客户的能力。

抽象的常用做法是:把「和客户系统打交道」的部分封装成统一的连接器接口,换客户只换实现,流程、评测、权限、审计全部复用:

from abc import ABC, abstractmethod

class RequirementSource(ABC):
    """统一接口:不同客户的需求系统各写一个实现"""
    @abstractmethod
    def fetch(self, req_id: str) -> dict: ...
    @abstractmethod
    def write_result(self, req_id: str, result: str) -> None: ...

class CustomerASource(RequirementSource):
    def fetch(self, req_id):
        data = http_get(f"https://a.example.com/api/req/{req_id}")
        return {"id": req_id, "title": data["name"], "body": data["desc"]}
    def write_result(self, req_id, result):
        http_post(f"https://a.example.com/api/req/{req_id}/comment", {"text": result})

class CustomerBSource(RequirementSource):
    def fetch(self, req_id):
        row = db_query("SELECT title, content FROM requirements WHERE id=%s", req_id)
        return {"id": req_id, "title": row["title"], "body": row["content"]}
    def write_result(self, req_id, result):
        db_execute("UPDATE requirements SET review=%s WHERE id=%s", result, req_id)

def build_source(customer: str) -> RequirementSource:
    return {"A": CustomerASource, "B": CustomerBSource}[customer]()

这样第二个、第三个客户的交付时间,会比第一个短得多。


七、一个具体例子:需求评审 Agent

阶段做什么
调研发现评审人要花大量时间读需求、查历史、对规范
目标评审前生成初审报告,评审耗时降低 40%,漏项率下降
摸底需求在需求管理系统,规范在文档库,需要读取权限,不能外传
POC加载评审规范 + 通过工具接口拉取需求 + 输出评审意见
评测写检查点:必需调用拉取需求,回答应覆盖完整性、可行性、风险点
上线先作为辅助建议,评审人确认后才写入,记录审计日志
推广培训评审人,统计采纳率,沉淀为通用评审模板

这类任务的评测可以用「检查点」的方式:必需检查点保证流程正确,可选检查点和回答检查点衡量质量。


八、FDE 做事的原则与常见坑

8.1 七条原则

  1. 价值优先于技术:能用简单方案解决,就不用复杂方案
  2. 先窄后宽:一个场景做透,再复制到其他场景
  3. 早交付、早反馈:每周都能给客户看到进展
  4. 用数据证明:评测结果是和客户沟通最有力的工具
  5. 管理预期:明确 AI 能做什么、不能做什么,承诺低一点,交付高一点
  6. 留好后路:人工兜底、可回滚、可审计
  7. 文档化与复用:把现场经验变成模板,避免每个客户从零开始

8.2 常见坑与对策

坑对策
需求一直变、范围蔓延阶段 2 写清范围,变更走评估
数据质量差,Demo 好看上线就崩阶段 3 提前摸底,用真实数据评测
一上来做全自主 Agent,效果不稳工作流为主,关键步骤人工确认
没有评测,改了不知道好坏第一周就搭评测集和回归
权限和安全最后才想一开始就设计权限边界
客户期望过高早期用数据对齐预期
做成一次性项目抽象组件,沉淀回产品

九、总结

  • FDE = 能写代码的客户现场负责人:既要工程和 AI 能力,也要业务理解、沟通和项目管理能力。
  • 技能栈以「广」为主:后端、数据、云、前端基础,加上 RAG、Agent、评测、安全,再加上需求挖掘和交付能力。
  • 做系统的节奏:调研、定标准、摸底、POC、评测迭代、生产化、推广、反馈产品,八步循环,每一步都围绕「客户价值」。
  • 评测是 FDE 的底牌:它既是迭代的依据,也是和客户建立信任的语言。
Logo

更多推荐