FDE Training · Module 01

01

全栈工程基础

模块 01 · 全栈工程基础（硬性门槛）

本章目标：把你从"会写 demo"拉到"能写可上线业务代码"。FDE 的 AI 能力再强，代码跑不起来白搭。这部分是大厂必查的硬门槛，没有讨价还价。

1.1 后端开发：Python 是主力（FastAPI/Flask）

1.1.1 为什么是 Python

客户侧 AI 项目里，Python 生态最全：FastAPI/Flask、vLLM、LangChain、向量库客户端全都有。
FDE 写的不是"算法脚本"，是能上线的业务服务：要登录、要鉴权、要限流、要返回统一错误码、要幂等重试。

1.1.2 用 FastAPI 写一个"能上线"的最小服务

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

app = FastAPI(title="合同审单服务", version="1.0.0")

class ReviewRequest(BaseModel):
    contract_id: str = Field(..., description="合同 ID")
    document: str = Field(..., min_length=1, max_length=100000)

class ReviewResponse(BaseModel):
    contract_id: str
    verdict: str          # approve / reject / need_manual
    score: float          # 置信度
    reasons: list[str]

@app.post("/api/review", response_model=ReviewResponse)
def review(req: ReviewRequest):
    # 真实项目这里会：鉴权 -> 落库 -> 调 LLM -> 存评估日志 -> 返回
    if not req.contract_id:
        raise HTTPException(status_code=422, detail="contract_id 不能为空")
    return ReviewResponse(contract_id=req.contract_id, verdict="need_manual",
                          score=0.5, reasons=["这是演示服务，请接入真实模型"])

能上线的标志：

用了 Pydantic 做输入校验（不是手动 if 一堆）。
有统一的响应模型。
加了统一错误处理（下面会讲）。
能连数据库、能写日志、能跑在 Docker 里（模块 02）。

1.1.3 几个"生产级"的小习惯（面试常考）

习惯	为什么
输入全部校验（Pydantic/校验库）	客户环境脏数据多，别让脏输入炸到模型
错误码统一	前端/客户 IT 好理解，`code + msg + data` 是最常见结构
幂等	客户网络重试，重复提交同一合同不能审两次、产生两笔费用
重试 + 退避	调用大模型偶尔超时，要有指数退避重试，不能裸调
结构化日志	出问题能从日志看链路，而不是满屏 print
单元测试	改代码不怕回归，客户要审计也有底气

# 一个简单的指数退避重试
import time, random

def call_with_retry(fn, retries=3, base=1.0):
    for i in range(retries):
        try:
            return fn()
        except Exception as e:
            if i == retries - 1:
                raise
            time.sleep(base * (2 ** i) + random.uniform(0, 0.5))

1.1.4 加分语言

Go / Java：客户内部系统常是 Java（Spring）或 Go，你至少能看懂、能对接。不需要精通，但面试问"你的主力语言外的工程经验"要有话说。

1.1.5 设计模式与 SOLID（面试高危题）

客户现场最怕一类代码：一个函数几百行、if-else 套三层，改一处炸一片。SOLID 是五条让你少返工的规矩，先记住最常用的两条——单一职责（S，Single Responsibility）：一个类/函数只干一件事；开闭原则（O，Open-Closed）：加功能别改老代码，靠扩展来加。

# 开闭原则：加新审单规则不碰老代码
class Rule:
    def match(self, doc): raise NotImplementedError
    def judge(self, doc): raise NotImplementedError

class AmountRule(Rule):
    def match(self, doc): return "金额" in doc
    def judge(self, doc): return "金额异常，转人工"

rules: list[Rule] = [AmountRule()]   # 以后加规则往这塞，别去改主流程
for r in rules:
    if r.match(doc):
        print(r.judge(doc))

设计模式就是这些原则落到具体场景的"套路"，FDE 最常用的两个是策略模式（把一堆 if-else 的审单规则替换成一组可插拔的策略类）和工厂模式（按合同类型创建不同处理器）。这套东西在 tech-cpp.html 里有成体系讲解——C++ 是讲设计模式最经典的载体，讲明白的道理跨语言通用。

1.2 API 设计：RESTful / gRPC

1.2.1 一套"客户看得懂"的 REST 规范

URL 用名词复数：/api/contracts、/api/reviews/{id}。
方法语义明确：GET 读、POST 建、PUT 整体更新、PATCH 部分更新、DELETE 删。
返回统一包一层：

{
  "code": 0,
  "msg": "ok",
  "data": { }
}

错误时 code 是非 0，例如 401 未登录、403 无权限、429 限流、500 服务内部错。

1.2.2 接口四大件（FDE 交付必配）

鉴权：至少 API Key 或 Bearer Token；对接客户身份系统（OAuth2/OIDC/SSO）放模块 07。
限流：保护后端和模型调用量。可用 Redis 滑动窗口或成熟中间件。
错误重试：客户端要对 429/5xx 做重试，服务端对下游（大模型）做重试。
幂等设计：请求带 Idempotency-Key，服务端按 key 去重，重复请求返回第一次的结果。

# 幂等：同一个 key 只处理一次
import hashlib

processed = set()  # 演示用，真实用 Redis 带 TTL

@app.post("/api/review")
def review(cls: str, req: ReviewRequest, idempotency_key: str):
    key = hashlib.sha256(idempotency_key.encode()).hexdigest()
    if key in processed:
        return {"code": 0, "data": {"cached": True}}
    processed.add(key)
    # ... 真实逻辑
    return {"code": 0, "data": {"cached": False}}

1.2.3 gRPC（知道是什么就好）

高频接口、内部服务间用 gRPC（更省流量、更快）；客户把他们的服务用 gRPC 暴露给你时你要能对接。
记住：gRPC 是二进制协议 + protobuf 定义，要能看 .proto 文件、能生成客户端。低频对接时也可以用 HTTP 网关规避。

1.3 数据库：SQL（MySQL/PostgreSQL）+ Redis + 向量库

1.3.1 SQL 熟练度是 FDE 的底线

客户的核心脑（数据）都在关系库里。你要会：

-- 连表统计：每月合同审核平均耗时
SELECT DATE_TRUNC('month', created_at) AS month,
       AVG(duration_sec), COUNT(*)
FROM reviews
WHERE status = 'done'
GROUP BY 1
ORDER BY 1;

另外三点拖家带口地把面试答好：

索引原理：B+ 树，为什么查 WHERE status='x' 有索引快。
慢查询排查：EXPLAIN，看全表扫描、回表、类型。
锁与事务：SELECT ... FOR UPDATE、悲观/乐观锁，理解脏读、不可重复读。

顺带把安全底线焊死：拼字符串拼 SQL 是 SQL 注入的温床，永远用参数化查询（Python 的 DB-API %s 占位、Java 的 PreparedStatement、ORM 框架都行），把用户输入当"数据"而不是"代码"传进去。

1.3.2 Redis：缓存 + 限流 + 幂等（高频）

缓存大模型结果：同一合同重复审直接读缓存，省钱又快。
限流：INCR + EXPIRE 做滑动窗口。
幂等：SETNX key value EX ttl。

import redis
r = redis.Redis(host="...", port=6379, db=0)

# 10 秒内最多 5 次
cur = r.incr("limit:user:123")
if cur == 1:
    r.expire("limit:user:123", 10)
if cur > 5:
    raise HTTPException(429, "请求过于频繁")

1.3.3 向量库（Chroma / Milvus / PGVector）

存文档分块的 embedding，供 RAG 检索（模块 04 深讲）。
现在记住一句话：向量库 = 按"语义相似度"查数据的库，SQL 用 = 精确匹配，向量库用 ≈ 相似度匹配。

1.3.4 缓存三大坑：穿透 / 击穿 / 雪崩

Redis 缓存不是"加一层就完事"，三个坑面试和现场都高频：

缓存穿透：查一个数据库里根本不存在的 key（比如合同 ID 不存在），每次都打穿缓存直连 DB。解法：布隆过滤器，或把"查不到"也缓存一个很短的空值。
缓存击穿：一个热点 key 过期瞬间，海量请求同时打到 DB。解法：热点 key 逻辑不过期、互斥锁重建。
缓存雪崩：一大批 key 同一时刻批量过期，DB 被瞬间压垮。解法：给过期时间加随机抖动（ttl * (1 + random())），错开过期。

import random
ttl = 60 * (1 + random.random())   # 过期时间加抖动，避免集体失效
r.setex(f"cache:contract:{cid}", int(ttl), value)

三个坑的成因、解法，以及和消息队列、注册中心的配合，系统性在 tech-middleware.html。

1.3.5 并发与锁：同一份合同别审两次

两个窗口同时审同一份合同，谁说了算？记住三类锁，按场景选：

悲观锁：上来就 SELECT ... FOR UPDATE 把行锁住，别人在旁等着。适合冲突多、要串行保护的场景。
乐观锁：表里带一个 version 字段，更新时 WHERE version = 旧值，更新影响 0 行就说明别人抢先了，你重试即可。适合读多写少。
分布式锁：服务有多个副本时，用 Redis 的 SET key val NX EX 10（谁 set 成功谁拿到锁），干完活再释放。前面 1.3.2 的幂等，本质就是靠它。

-- 乐观锁：version 对不上 = 别人已改，重试
UPDATE reviews SET verdict = 'approve', version = version + 1
WHERE id = 123 AND version = 3;   -- 受影响 0 行，说明版本已变

锁、原子操作、多线程竞争在 tech-cpp.html 的并发章节讲得最透（RAII、互斥锁、原子变量，把"为什么加锁、锁错在哪"讲到底）；事务隔离级别、行锁在 tech-database.html 补。

顺带补一句数据库工程化常识：别每个请求都现开一条数据库连接，用连接池复用连接、限制上限（客户内网 DB 连接数往往很紧张）；连接池、JDBC、声明式事务在 tech-javaee.html 有系统讲解。

1.4 前端：够用即可（TS/JS + React）

FDE 不需要是顶级前端，但要：

能改写页面：客户要改管理后台、调试面板。
能对接接口：看懂 fetch/axios、把后端数据渲染出来。
能做 POC 原型：给客户演示的全栈 demo。

一个够用的 React 组件长相：

import { useEffect, useState } from "react";

export default function ReviewList() {
  const [items, setItems] = useState<{id: string; verdict: string}[]>([]);
  useEffect(() => {
    fetch("/api/reviews").then(r => r.json()).then(j => setItems(j.data ?? []));
  }, []);
  return (
    <ul>
      {items.map(it => <li key={it.id}>{it.id} → {it.verdict}</li>)}
    </ul>
  );
}

要点：看得懂 TypeScript 类型、会用 useState/useEffect、会调接口就够入门；别在样式上内卷。

1.5 脚本与自动化：Shell / PowerShell

客户现场大量重复操作要自动化：

自动化部署：一键拉代码、装依赖、起服务。
巡检：定时检查服务健康、磁盘、进程。
数据 ETL：把客户 CSV/Excel 洗进数据库。

#!/usr/bin/env bash
# 简单健康巡检
set -euo pipefail
url="http://127.0.0.1:8080/api/health"
code=$(curl -s -o /dev/null -w "%{http_code}" "$url")
echo "health status: $code"
if [ "$code" != "200" ]; then
  echo "service unhealthy, restarting..." >&2
  systemctl restart my-service
fi

另外：Git 熟练、PR 评审、代码规范是基本功，别在这上面丢分。

1.6 工程素养：日志、链路、异常、测试、压测

FDE 交付的东西要可排查、可回归、可量化：

结构化日志：JSON 格式，含 request_id、耗时、是否成功。
链路追踪：一次请求从入口到调大模型，靠 request_id 把所有步骤串起来。
异常兜底：模型挂了给降级响应、数据库连不上给友好错误，别让用户看到满屏堆栈。
单元测试：核心函数（分块、解析、校验）要有测试，覆盖率优先保住核心逻辑。
性能压测：用 ab/locust 简单压一下，知道"多少并发会崩"。

import logging, json, time, uuid

log = logging.getLogger("review")
req_id = uuid.uuid4().hex[:12]

def log_event(name, **kw):
    entry = {"req_id": req_id, "event": name, "ts": time.time(), **kw}
    log.info(json.dumps(entry, ensure_ascii=False))

1.7 模块练习

用 FastAPI 写一个 /api/health 和一个带 Pydantic 校验的 POST 接口，跑通返回统一 {code,msg,data}。
给你的接口加：Redis 限流（10 秒 5 次）+ 一个 Idempotency-Key 幂等。
写一个匿名 EXPLAIN 来分析这条 SQL 为什么慢，并补一个合适索引：SELECT * FROM reviews WHERE user_id=1 AND status='done' ORDER BY created_at DESC。
写一个 bash 巡检脚本：检查服务健康，非 200 就重启并写日志。
给你接口的调用大模型部分加"指数退避重试"，并用单元测试覆盖重试成功/失败两条路径。

1.8 本章面试题

"你怎么设计一个生产级的 API？" → 答：REST 语义、统一返回 {code,msg,data}、Pydantic 输入校验、Bearer/API Key 鉴权、Redis 限流、幂等键、结构化日志带 request_id、异常统一兜底、对下游模型指数退避重试。
"解释一下 SQL 慢查询排查流程。" → 答：先看 EXPLAIN 确认是全表扫描还是走索引 → 看是否缺联合索引 / 索引失效（函数包裹列、隐式类型转换）→ 补索引或改写 SQL → 压测确认耗时下降。
"Redis 你在客户项目里一般怎么用？" → 答：缓存大模型结果（省 token、降延迟）、限流（INCR+EXPIRE）、幂等（SETNX）、分布式锁、会话/状态缓存。
"'能上线'和'写 demo'的代码本质区别是什么？" → 答：demo 只求能跑；上线要能扛：校验、鉴权、限流、重试幂等、日志链路、异常兜底、测试、压测，还要能排查、能审计。
"让你对接一个客户内部用 gRPC 的老系统，你会怎么做？" → 答：先要 .proto 定义和数据字段说明 → 生成客户端 → 用网关/适配层把 gRPC 转成团队熟悉的 HTTP 接口 → 加上鉴权、超时、重试 → 在隔离环境联调通过后再接生产。

1.9 小结

后端 Python + FastAPI 是主力；写的是能上线的代码，不是 demo。
API 四大件：鉴权、限流、重试、幂等；统一返回结构。
SQL + Redis + 向量库是数据三件套；会查慢查询、会用缓存。
前端够用即可（TS/React/调接口）；脚本用 Shell；工程素养是底线。
下一章进入云原生与基础设施——"客户现场交付"的核心战场。