编程与工程基础 Engineering Foundations

阶段总览

1. Python 工程能力

1.1 现代 Python：从能跑 → 可维护

AI 代码的最大反模式是「一个 3000 行 notebook 直接上生产」。2026 年面试官看简历时会直接问：你的训练/推理代码有类型标注吗？有单测吗？配置和密钥怎么管？这三个问题的答案，比你的模型指标更能说明工程素养。

pythonfrom dataclasses import dataclass
from typing import Protocol, Iterator
from pydantic import BaseModel, Field
from pydantic_settings import BaseSettings   # 运行时校验 + 环境变量注入

class Settings(BaseSettings):
    model_name: str = 'Qwen/Qwen3-8B'
    max_new_tokens: int = Field(512, ge=1, le=32768)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    api_key: str                        # 缺失时启动即报错，而不是跑到一半才炸
    class Config:
        env_file = '.env'
        env_prefix = 'HAMAULS_ORION_'

class LLM(Protocol):                    # 结构化子类型：只依赖接口，不依赖实现
    """为什么用 Protocol 而不是抽象基类？
       Python 是鸭子类型，Protocol 让你能对「任何长得像 LLM 的对象」编程，
       无需让第三方类继承你的基类 —— 这大幅降低了替换供应商的成本。
       在 Hamauls Orion 里，NullPushChannel / vLLM / 商业 API 都只需满足这个接口。"""
    def chat(self, messages: list[dict], **kw) -> str: ...

@dataclass(frozen=True, slots=True)
class Sample:                           # 不可变 + 省内存，评测集常用
    prompt: str
    reference: str
    meta: dict

def chunked(it: Iterator[str], n: int) -> Iterator[list[str]]:
    """惰性分批：处理 1 亿条语料时，绝不能先 list() 再切"""
    buf: list[str] = []
    for x in it:
        buf.append(x)
        if len(buf) >= n:
            yield buf; buf = []
    if buf: yield buf

• 类型标注 + mypy / pyright：动态语言的类型标注不会在运行时生效，但能在 CI 里拦掉大量低级错误，更是现代 IDE 自动补全与重构的前提。代价是几天的学习，收益是长期的。
• Pydantic 校验边界：所有来自外部（用户输入、LLM 输出、API 响应、数据库反序列化）的数据都必须在边界处校验。LLM 的输出本质是「不可信输入」，这是新手最常忽略的一点。
• 结构化日志而非 print：用 logging 或 structlog，训练与推理都要能按 request_id / trace_id 串联全链路。日志要能直接进 ELK / Loki 查询，而不是靠 grep 堆文件。
• 异常分层的思路：区分「可重试」（限流、超时、网络抖动）与「不可重试」（鉴权失败、参数非法、schema 不匹配），只有前者才进重试逻辑。这个分类会在第 4 阶段的可靠性设计里被正式工程化。
• __slots__ 与 frozen：数据密集的对象（一个样本、一个检索结果）用 slots=True 能省 30–50% 内存；frozen=True 避免意外修改导致的难查 bug。

python# 可重试 vs 不可重试 —— 这是所有 AI 服务的通用骨架
import asyncio, random, logging
from typing import Awaitable, Callable, TypeVar

log = logging.getLogger(__name__)
T = TypeVar('T')

class RateLimitError(Exception): ...
class AuthError(Exception): ...

RETRYABLE = (asyncio.TimeoutError, ConnectionError, RateLimitError)
FATAL     = (AuthError, ValueError)          # 明确列出，避免 except Exception 一把抓

async def with_retry(fn: Callable[[], Awaitable[T]], tries: int = 4,
                     base: float = 0.8, cap: float = 30.0) -> T:
    for i in range(tries):
        try:
            return await fn()
        except FATAL:
            raise                            # 不可重试：立刻抛，不浪费配额
        except RETRYABLE as e:
            if i == tries - 1:
                raise
            # 指数退避 + 抖动（jitter）：避免大量客户端同步重试造成惊群
            delay = min(cap, base * (2 ** i)) * (0.5 + random.random() * 0.5)
            log.warning('retry %d/%d after %.2fs: %s', i + 1, tries, delay, e)
            await asyncio.sleep(delay)

# 注意这里没有 except Exception —— 让未知异常直接冒泡。
# 「宽泛捕获」会掩盖真正的 bug，是最常见的技术债来源。

python# Pydantic v2 在「系统边界」拦截脏数据 —— LLM 输出 = 不可信输入
from pydantic import BaseModel, Field, field_validator
from typing import Literal

class Answer(BaseModel):
    answer: str = Field(min_length=1, max_length=8000)
    cites: list[int] = Field(default_factory=list, max_length=20)
    route: Literal['direct', 'rag', 'tool'] = 'direct'

    @field_validator('cites')
    @classmethod
    def normalize_cites(cls, v):
        return sorted(set(v))          # 归一化：去重 + 排序，下游不必再防御

obj = Answer.model_validate_json('{"answer": "RoPE 用旋转编码相对位置", "cites": [3, 1, 3]}')
print(obj.cites, obj.route)
# 预期输出: [1, 3] direct        ← 3 被去重、顺序被规范化

try:
    Answer.model_validate_json('{"answer": ""}')
except Exception as e:
    print(type(e).__name__)
# 预期输出: ValidationError       （min_length=1 拦截空答案，不会静默通过）

1.2 并发与异步：AI 代码的性能命门

LLM 调用是典型的 I/O 密集任务：99% 的时间在等网络。用同步 for 循环调 1000 条数据，耗时是并发的几十倍。这不是「优化技巧」，而是「能不能在今晚跑完评测集」的问题。

pythonimport asyncio, aiohttp, json

async def eval_one(session, sem, item):
    async with sem:                             # 限流：保护对端，也保护自己不被封
        async with session.post(URL, json=item) as r:
            return await r.json()

async def run(items, concurrency=16):
    sem = asyncio.Semaphore(concurrency)
    async with aiohttp.ClientSession() as s:
        tasks = [eval_one(s, sem, it) for it in items]
        return await asyncio.gather(*tasks, return_exceptions=True)

# 大评测集要边跑边落盘，避免中途失败全部重来 —— Hamauls Orion M1 的核心组件
async def run_streaming(items, out_path, concurrency=16):
    sem = asyncio.Semaphore(concurrency)
    done = 0
    async with aiohttp.ClientSession() as s, open(out_path, 'a') as f:
        for coro in asyncio.as_completed([eval_one(s, sem, it) for it in items]):
            try:
                f.write(json.dumps(await coro, ensure_ascii=False) + '\n')
                f.flush()                       # 断点续跑的基础（OS 缓冲会吞数据）
            except Exception as e:
                # 单条失败不能中断整个流水线 —— 记录并继续
                f.write(json.dumps({'error': str(e)}) + '\n'); f.flush()
            done += 1
            if done % 100 == 0:
                print(f'{done}/{len(items)}')

# 断点续跑：启动时读取已完成的行，跳过已处理的（这是工程上必须的能力）
async def resume(items, out_path, **kw):
    import os
    if os.path.exists(out_path):
        with open(out_path) as f:
            done = sum(1 for _ in f)
        items = items[done:]
        print(f'从第 {done} 条续跑，剩余 {len(items)} 条')
    await run_streaming(items, out_path, **kw)

python# 并发数不是越大越好：实测「并发 — 吞吐」曲线，找拐点
import asyncio, time, random

async def one(sem, lat=0.10):
    async with sem:
        await asyncio.sleep(lat * (0.8 + 0.4 * random.random()))
        return 1

async def bench(n_items=200, concurrency=16, lat=0.10):
    sem = asyncio.Semaphore(concurrency)
    t0 = time.perf_counter()
    await asyncio.gather(*[one(sem, lat) for _ in range(n_items)])
    dt = time.perf_counter() - t0
    return n_items / dt                    # QPS

async def main():
    for c in (1, 8, 32, 128, 512):
        print(f'concurrency={c:3d}  qps={await bench(200, c):8.1f}')

asyncio.run(main())
# 预期输出（单条纯等待约 0.1s）：
# concurrency=  1  qps=    10.0
# concurrency=  8  qps=    79.5
# concurrency= 32  qps=   315.2
# concurrency=128  qps=  1260.9
# concurrency=512  qps=  1800.0   ← 事件循环与调度开销开始压住收益，趋于饱和

用排队论定并发上限：Little 定律 L = λ · W（系统内平均在途数 = 到达率 × 平均逗留时间）。若单条请求平均耗时 W 秒、目标吞吐 λ QPS，则稳态所需并发约 L = λ · W。例如 W=2s、想跑 50 QPS，需要约 100 并发；因为长尾（P99 常是 P50 的 5–20 倍），实际要留 2–3 倍余量。而吞吐随并发增长会饱和：并发超过对端承载后排队使 W 上升，λ 反而下降（重试放大）。这就是「并发调大反而变慢」的机理。

1.3 工程化：包管理、项目结构、测试与可复现

textHamauls Orion 的仓库布局（这一层的设计会一直用到第 17 阶段）

hamauls-orion/
├── pyproject.toml          # 项目元数据 + 依赖（uv 管理）
├── uv.lock                 # 锁文件，必须提交到 git
├── README.md
├── Makefile                # make dev / make test / make build 统一入口
├── configs/                # 配置（hydra 风格），不入 git 的只有 secrets
│   ├── base.yaml
│   ├── train/sft.yaml
│   └── serving/vllm.yaml
├── src/hamauls_orion/              # 源码放在 src/ 下（避免「当前目录恰好能 import」的假象）
│   ├── core/               # 配置、日志、异常、通用类型
│   ├── data/               # 数据加载、清洗、版本管理
│   ├── models/             # 模型封装（Protocol 定义在这里）
│   ├── index/              # 检索索引（第 3 阶段的产出落这里）
│   ├── rag/                # 检索增强管线（第 13 阶段）
│   ├── agent/              # Agent 与工具（第 14 阶段）
│   ├── serving/            # 网关、超时、重试、限流（第 4、16 阶段）
│   └── evals/              # 评测与门禁（第 15 阶段）
├── tests/
│   ├── unit/               # 纯函数单测，快
│   ├── integration/        # 需要外部依赖，标记为 slow
│   └── snapshots/          # prompt / schema 快照
├── scripts/                # 一次性脚本与入口
├── notebooks/              # 探索性分析，不允许被 src/ import
└── docker/
    ├── Dockerfile
    └── docker-compose.yaml

• 包管理：新项目用 uv（2026 年事实上的快标准）或 poetry；锁文件必须提交。绝不要在服务器上 pip install 一堆未锁版本的包——这是「上周还能跑，今天跑不了」的头号原因。
• 为什么代码放 src/：如果源码在仓库根目录，import hamauls_orion 会在「当前工作目录」生效，导致本地能跑、安装后崩溃。src/ 布局强制你以「已安装包」的方式测试，能提前发现打包问题。
• 配置分层：默认值写代码里，环境差异用环境变量 / Profile 覆盖，密钥永不出现在仓库（用 .env + .gitignore，生产用密钥管理服务）。
• 测试三件套（AI 项目最容易漏后面两类）：① 纯函数单测（pytest，毫秒级，必须占大多数）；② LLM 输出断言——不比对全文，而是断言结构（能被 schema 解析）、关键字段存在、以及用 LLM-as-Judge 打分是否达标；③ 端到端 smoke test（跑通一条最短链路，验证接线正确），标记为 @pytest.mark.slow 不进常规 CI。
• 可复现的四个抓手：固定随机种子（含 numpy / torch / python random 三处）、记录数据版本哈希、记录模型版本、把实验参数写进配置而不是散落在脚本里。

python# tests/unit/test_pipeline.py —— AI 项目里最该写的三类测试
import pytest
from pydantic import ValidationError

def test_schema_validation():
    """LLM 输出必须能被 schema 校验，脏数据要降级而不是崩"""
    assert parse_or_default('{"a": 1}', schema=MySchema) is not None
    assert parse_or_default('抱歉我无法回答', schema=MySchema) is None   # 降级路径

def test_prompt_snapshot():
    """Prompt 变更必须被看见：快照测试防止顺手改坏"""
    assert build_prompt(SAMPLE_INPUT) == open('tests/snapshots/p1.txt').read()

def test_gradable_llm_output():
    """不比对全文（模型输出必然变化），而是断言结构与关键约束"""
    out = call_model('用 JSON 输出用户姓名和年龄')
    parsed = json.loads(out)                     # 结构必须可解析
    assert set(parsed) >= {'name', 'age'}
    assert isinstance(parsed['age'], int)

@pytest.mark.slow
def test_e2e_smoke():
    """跑通一条最短链路，验证接线正确（用最小模型 / mock）"""
    assert run_pipeline('你好')['answer']

toml# pyproject.toml —— 现代 Python 项目的最小可信配置（uv / ruff / mypy / pytest 一处集中）
[project]
name = "hamauls-orion"                  # 发行名可用连字符
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["pydantic>=2.7", "httpx>=0.27", "polars>=1.0", "structlog>=24.1"]

[project.scripts]
hamauls-orion = "hamauls_orion.cli:main"   # 左侧命令名可连字符，右侧模块名下划线

[dependency-groups]
dev = ["pytest>=8.0", "pytest-cov>=5.0", "ruff>=0.6", "mypy>=1.10", "pre-commit>=3.7"]

[tool.ruff]
line-length = 100
target-version = "py312"

[tool.ruff.lint]
select = ["E", "F", "I", "B", "UP", "SIM"]     # I=isort, B=bugbear, UP=pyupgrade

[tool.mypy]
strict = true
warn_unreachable = true

[tool.pytest.ini_options]
markers = ["slow: 需要外部依赖的端到端测试"]
addopts = "-q -m 'not slow'"

1.4 调试与性能剖析：先测量再优化

这一节与第 2 阶段的性能分析呼应，但聚焦在Python 层的定位手段。原则只有一条：不要猜，要测。

bash# ① 找到热点函数（cProfile：函数级调用次数与自耗时）
python -m cProfile -o prof.out -s cumtime scripts/run_eval.py
python -c "import pstats; pstats.Stats('prof.out').sort_stats('cumtime').print_stats(20)"

# ② 附加到正在运行的进程（生产环境排查，不重启）
py-spy dump --pid <PID>              # 立刻打印所有线程的调用栈 ← 排查「卡住了」
py-spy top --pid <PID>               # 实时看哪个函数在烧 CPU
py-spy record -o profile.svg --pid <PID>   # 生成火焰图

# ③ 逐行耗时（找到热点函数后用）
kernprof -l -v scripts/hot_function.py     # line_profiler

# ④ 内存泄漏排查
tracemalloc / objgraph / memray run script.py
memray flamegraph memray.bin               # 内存火焰图，定位增长点

# ⑤ 最朴素但最有效的一招：手工计时（粗粒度先定位到模块）
python -c "
import time
t = {}
def tick(k):
    now = time.perf_counter()
    if k in t: print(f'{k}: {now - t[k]:.3f}s')
    t[k] = now
tick('start'); load_data(); tick('load')
retrieve(); tick('retrieve'); generate(); tick('generate')
"

• 顺序永远是：粗粒度计时 → 函数级 profiler → 行级 → 底层（GPU/网络）。跳过前两步直接上高级工具，通常是在优化不重要的地方。
• py-spy 是生产环境最实用的工具：低开销、无需改代码、能附加到运行中的进程。当线上服务「卡住但没报错」时，py-spy dump 能在 5 秒内告诉你它卡在哪一行。
• 小心 profiler 带来的失真：cProfile 会给每次函数调用加开销，所以它夸大了「调用次数多的小函数」的耗时。对照 time.perf_counter() 的手工计时来校准判断。
• 字符串拼接是隐藏热点：大循环里 s += x 在 Python 里是 O(n²)。改成 join(list)（空串 join）往往能快几十倍；同理，日志格式化要用 log.info("%s", x) 而不是 log.info(f"{x}")——后者即使日志级别不输出也会先做字符串拼接。

python# 两个立竿见影的微优化（在内层循环里差异巨大）
import logging, time
log = logging.getLogger(__name__)

# ❌ f-string 即使日志被过滤掉也已经付出了格式化成本
for x in big_list:
    log.debug(f'processing {x}')          # 慢

# ✅ 延迟格式化：只有真的要输出时才格式化
for x in big_list:
    log.debug('processing %s', x)         # 快

# ❌ 字符串累加 O(n²)
out = ''
for tok in tokens: out += tok             # 每次都在重新分配

# ✅ join 一次性分配
out = ''.join(tokens)

# 实测对比（10 万个短字符串）：
#   += 累加: 约 0.35s
#   join   : 约 0.002s      → 快 100 倍以上

python# 用 timeit 把「优化前后」量化，避免凭感觉
import timeit, random

tokens = [str(random.random()) for _ in range(100_000)]

def bad():
    s = ''
    for x in tokens:
        s += x
    return s

t_join = timeit.timeit(lambda: ''.join(tokens), number=100) / 100
t_add  = timeit.timeit(bad, number=20) / 20
print(f'join={t_join*1000:.2f}ms  plus={t_add*1000:.2f}ms  speedup={t_add/t_join:.0f}x')
# 预期输出（10 万字符串，单次）：
# join=1.30ms  plus=210.00ms  speedup=160x   ← 数量级差距，不是「差不多」

1.5 动手练习与自测（Python 工程）

1. 在本机用 timeit 对比 += 与 join 处理 10 万个短字符串，报出倍数并解释复杂度差异。判据：倍数通常 ≥100x；+= 每次重新分配整串，是 O(n²)，join 先算总长再一次性分配，是 O(n)。
2. 用 Little 定律估算所需并发：若单条请求平均耗时 W=1.5s、目标吞吐 λ=40 QPS，需要多少并发？为什么还要再留余量？参考答案：约 40 × 1.5 = 60 并发；因 P99 长尾常是 P50 的 5–20 倍，实际需 2–3 倍余量，否则尾延迟爆炸。
3. 给一段在 async 函数里直接调用 requests.get 与 time.sleep 的代码，指出问题与两种修法。判据：两者都会阻塞事件循环，使全部并发退化为串行；修法一：await asyncio.to_thread(...)；修法二：换成 httpx.AsyncClient / asyncio.sleep。
4. 把 6 种典型异常分成「必须重试 / 绝不重试」两类，并说明退避策略。参考答案：重试=429 限流、超时、连接错误 / 5xx（指数退避 + 抖动，上限 30s，次数 2–4）；不重试=401/403 鉴权、400 参数非法 / schema 不符（重试只浪费配额，应立刻失败并告警）。
5. 为什么 uv.lock 必须提交进 git？不提交会发生什么？判据：锁文件固定了每个依赖的精确版本与哈希，保证「今天能跑、下周还能跑」；不提交则不同机器 / 不同时间解析到不同版本，产生「薛定谔的可复现」。
6. 线上 Python 服务「卡住但没报错」，写出三步定位命令。参考答案：① py-spy dump --pid 看所有线程调用栈（最快判断卡在哪一行）；② ps aux | grep 确认进程与 CPU；③ strace -p -e trace=network 看系统调用。若栈停在同步 I/O 或锁上，基本即锁定原因。

2. TypeScript：AI 应用的交付层

2.1 为什么 AI 工程师也要会一点前端

2026 年的 AI 产品绝大多数以「对话 + 流式输出 + 工具调用可视化」的形态交付。而流式、SSE、断线重连、消息状态机、把模型的 tool_call 渲染成可交互卡片——这些都在前端。会用 TypeScript 意味着你能独立把 demo 变成产品，而不是卡在「等前端排期」。

typescript// 消费 SSE 流：AI 应用最小骨架（前端）
type Chunk = { delta?: string; tool_call?: unknown; usage?: unknown };

async function* streamChat(messages: Message[], signal?: AbortSignal) {
  const res = await fetch('/api/chat', {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({ messages }),
    signal,                                   // 支持「停止生成」
  });
  if (!res.ok || !res.body) throw new Error(await res.text());

  const reader = res.body.getReader();
  const decoder = new TextDecoder();
  let buf = '';
  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    buf += decoder.decode(value, { stream: true });
    const parts = buf.split('\n\n');        // SSE 以空行分帧
    buf = parts.pop() ?? '';
    for (const part of parts) {
      if (part.startsWith(':')) continue;     // 注释行 = 心跳，跳过
      const line = part.replace(/^data:\s*/, '');
      if (line === '[DONE]') return;
      yield JSON.parse(line) as Chunk;        // {delta} | {tool_call} | {usage}
    }
  }
}

// 消费端：把一个「流」变成「可取消 + 可重试 + 状态明确」的 UI
async function send(text: string, setState: (s: MsgState) => void, ctrl: AbortController) {
  setState({ phase: 'loading', text: '' });
  try {
    for await (const c of streamChat(history, ctrl.signal)) {
      setState(prev => ({ phase: 'streaming', text: prev.text + (c.delta ?? '') }));
    }
    setState(prev => ({ ...prev, phase: 'done' }));
  } catch (e) {
    if ((e as Error).name === 'AbortError') setState(prev => ({ ...prev, phase: 'stopped' }));
    else setState(prev => ({ ...prev, phase: 'error', error: String(e) }));
  }
}

• 必须懂的概念：fetch + ReadableStream、SSE 分帧（空行分隔、data: 前缀、注释行心跳）、AbortController、乐观更新、虚拟列表（长对话性能）、以及「流式渲染时的 markdown 增量解析」。
• 推荐的栈：React + TypeScript + TanStack Query；服务端流式用 Next.js Route Handler 或独立 FastAPI。
• 不要重复造轮子：Vercel AI SDK 这类库已封装了流式与工具调用协议，先学会用，再学会它封了什么。但一定要手写一次原始 SSE 消费，否则出问题时你无法定位。
• 边界校验：前端所有外部数据（含模型输出）也要校验；后端已是不可信输入，前端同样——尤其是当你要把模型输出渲染成 HTML 时（XSS 风险），必须走 markdown 安全渲染而不是 innerHTML。

typescript// 生产级 SSE 消费：断线重连 + 续传 + 指数退避（对应 2026 的 AI 流协议）
async function* connect(url: string, signal: AbortSignal) {
  let lastId: string | null = null;
  let attempt = 0;
  while (!signal.aborted) {
    const headers: Record<string, string> = { Accept: 'text/event-stream' };
    if (lastId) headers['Last-Event-ID'] = lastId;   // 让服务端从断点续发
    try {
      const res = await fetch(url, { headers, signal });
      if (!res.ok || !res.body) throw new Error(String(res.status));
      attempt = 0;                                    // 连上就重置退避
      for await (const evt of parseSSE(res.body)) {
        if (evt.id) lastId = evt.id;
        yield evt;
      }
    } catch (e) {
      if (signal.aborted) return;
      attempt += 1;
      const wait = Math.min(30_000, 2 ** attempt * 500) * (0.5 + Math.random() * 0.5);
      await new Promise(r => setTimeout(r, wait));    // 指数退避 + 抖动
    }
  }
}
// 关键点：EventSource 只能 GET、无法自定义 header / body，
// 生产里普遍用 fetch + ReadableStream 手写，而不是原生 EventSource。
// 若中途 reader 未读满就切换，务必要 reader.cancel() 释放连接。

2.2 动手练习与自测（TypeScript 交付层）

1. 写一个最小 SSE 消费循环，遇到 \n\n 分帧、跳过注释行心跳、识别 [DONE]。判据：缓冲未闭合的尾帧、注释行以 : 开头跳过、data: 前缀剥离后再 JSON.parse。
2. 为什么生产里多用 fetch + ReadableStream 而不是原生 EventSource？参考答案：EventSource 只能发起 GET、不能带 Authorization 等自定义 header、不能发 POST body、错误重连能力有限；LLM 对话要 POST 且要鉴权，因此手写。
3. 实现「停止生成」，并说明服务端要做什么。判据：前端用 AbortController 把 signal 传给 fetch；服务端需在连接断开 / 取消时终止推理（否则继续占用 GPU 生成无人消费的 token）。
4. SSE 的断点续传需要客户端与服务端各做什么？判据：服务端每条事件带 id:；客户端重连时带 Last-Event-ID 头；服务端据此从该 id 之后续发，避免重复与丢失。
5. 把模型输出渲染进页面有哪些安全与性能坑？判据：安全——XSS，禁止 innerHTML 直插，用安全 markdown 渲染（如 DOMPurify 或多层白名单）；性能——避免每个 token 触发全量重渲染（用增量 markdown 解析 + 虚拟列表）。

3. 数据工程与 SQL：被低估的胜负手

3.1 SQL 进阶：窗口函数、CTE 与 JSON

2026 年最稀缺的不是会调模型的人，而是能把数据搞干净的人。训练数据、SFT 指令集、评测集、RAG 语料——全都先经过数据管线。多数学到一半放弃的 RAG 项目，死因是数据而不是模型。

sql-- 构造评测集：从生产日志里挑出「有参考改写」且「分布不偏」的样本
WITH ranked AS (
  SELECT
    q.id, q.question, q.answer, q.rewritten_at, q.intent,
    -- 窗口函数①：按意图分区，算最近 30 条的平均质量
    AVG(CASE WHEN f.helpful THEN 1.0 ELSE 0.0 END)
      OVER (PARTITION BY q.intent ORDER BY q.rewritten_at
            ROWS BETWEEN 29 PRECEDING AND CURRENT ROW) AS recent_quality,
    -- 窗口函数②：按意图分区随机排序，用于分层抽样
    ROW_NUMBER() OVER (PARTITION BY q.intent ORDER BY RANDOM()) AS rn
  FROM queries q
  LEFT JOIN feedback f USING (id)
  WHERE q.created_at >= CURRENT_DATE - INTERVAL '30 days'
    AND q.answer IS NOT NULL
)
SELECT * FROM ranked
WHERE recent_quality >= 0.7 AND rn <= 50     -- 每个意图类均匀取样（避免分布偏斜）
ORDER BY intent, rn;

-- JSON 字段查询（模型元数据、工具调用参数常存在 JSON 列里）
SELECT
  payload->>'model'                                   AS model,
  (payload->'usage'->>'completion_tokens')::int       AS out_tokens,
  COUNT(*)                                            AS calls,
  SUM((payload->'usage'->>'total_tokens')::int)       AS tokens
FROM llm_calls
WHERE created_at > now() - INTERVAL '1 day'
GROUP BY 1, 2
ORDER BY tokens DESC;

• 窗口函数的三个必会模式：① 排名（ROW_NUMBER / RANK / DENSE_RANK 的区别要清楚）；② 累计与滑动（ROWS BETWEEN ... AND ...，注意默认帧是 RANGE 而非 ROWS，这是个经典陷阱）；③ 前后对比（LAG / LEAD，用于算留存、转化、变化率）。
• CTE 让复杂查询可读：把「筛选 → 聚合 → 排名 → 采样」拆成多段 CTE，比嵌套子查询清晰得多。现代查询优化器对 CTE 的优化已经很好，不必因为性能而拒绝使用。
• 分层抽样比随机抽样更重要：随机抽样会让高频意图淹没有效样本。按业务维度（意图 / 长度 / 语言 / 难度）分区后各取 N 条，才得到有代表性的评测集。这是数据工作的核心手艺。
• 评测集与训练集隔离：同一来源的样本绝不能横跨两边，否则所有指标都是幻觉。更严格的做法是按「用户 / 时间 / 文档」维度做隔离，而不只是按行随机切分。

sql-- 训练/评测隔离：按「用户」哈希切分，防止同源泄漏
WITH base AS (
  SELECT id, user_id, created_at,
         ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY created_at) AS user_seq
  FROM samples
  WHERE created_at < now()
)
SELECT id, user_id, created_at,
  CASE WHEN user_id % 10 = 0 THEN 'test'
       WHEN user_id % 10 = 1 THEN 'valid'
       ELSE 'train' END AS split      -- 同一用户只进一个集合，杜绝泄漏
FROM base
WHERE user_seq <= 20;                 -- 每用户最多取 20 条，抑制头部用户权重

-- 泄漏自检（必须返回 0 行，否则说明同一用户横跨多个 split）
SELECT user_id FROM (
  SELECT user_id, COUNT(DISTINCT split) AS n FROM dataset_view GROUP BY 1
) t WHERE n > 1;

-- 上亿行也别忘了索引与执行计划：EXPLAIN ANALYZE 看是否走了 Seq Scan
EXPLAIN ANALYZE SELECT count(*) FROM samples WHERE created_at > now() - INTERVAL '7 days';

3.2 pandas → polars 与数据质量四查

python# pandas → polars：数据量上到千万行以后的速度差距是数量级的
import polars as pl

df = pl.scan_ndjson('raw/*.jsonl')            # lazy：先规划再执行，可下推谓词
clean = (
    df.filter(pl.col('text').str.len_chars() > 50)
      .with_columns(
          pl.col('text').str.replace_all(r'\s+', ' ').str.strip_chars().alias('text'),
          pl.col('text').str.len_chars().alias('n_chars'),
      )
      .unique(subset=['text'], keep='first')   # 精确去重
      .collect()
)
# 近重复（MinHash）与语义重复（embedding 聚类）要另做，别指望 unique() 解决
# —— 这一块会在第 3 阶段的概率数据结构里正式实现

# 数据质量四查：每次数据更新都跑一遍，输出可存档的报告
def quality_report(path: str) -> dict:
    df = pl.scan_ndjson(path)
    return {
        'total':     df.select(pl.len()).collect().item(),
        'null_rate': df.select([pl.col(c).is_null().mean().alias(c) for c in COLS]).collect().to_dicts()[0],
        'len_p50':   df.select(pl.col('text').str.len_chars().median()).collect().item(),
        'len_min':   df.select(pl.col('text').str.len_chars().min()).collect().item(),
        'lang_dist': df.group_by(pl.col('lang')).len().collect().to_dicts(),
        'dup_exact': df.select(pl.len()).collect().item()
                     - df.unique(subset=['text']).select(pl.len()).collect().item(),
        'empty_text': df.filter(pl.col('text').str.strip_chars().str.len_chars() == 0)
                        .select(pl.len()).collect().item(),
    }
# 把这四个维度（空值 / 长度分布 / 语言分布 / 重复率）固化成一个脚本，
# 每次数据版本更新都跑，把报告存进 docs/data/ 作为数据卡的一部分。

• polars 为什么快：Rust 实现、多核并行、惰性执行与查询优化（谓词下推、投影下推）、Apache Arrow 列式内存布局（对向量化与 cache 友好——这正好呼应第 2 阶段的存储层次）。
• 迁移的心智转变：从 pandas 的「索引 + 就地修改」变成 pl 的「表达式 + 不可变链式」。一旦适应，代码更短、更快、更少 bug。
• 数据血缘与版本：用 DVC 或简单的「目录 + 内容哈希」记录。训练要能回答「这份权重是哪个数据版本训出来的」。没有数据版本，你就无法复现任何实验。
• 数据卡（datasheet）：记录数据来源、许可证、清洗步骤、已知偏差与适用边界。这在第 9 阶段的领域微调里是硬性交付物——它是「负责任的数据使用」的最低要求，也是面试里的加分项。

python# pandas vs polars：同一份约 500 万行文本清洗的实测差距
import time, pandas as pd, polars as pl

t0 = time.perf_counter()
pdf = pd.read_json('raw.jsonl', lines=True)
pdf = pdf[pdf['text'].str.len() > 50]
pdf['text'] = pdf['text'].str.replace(r'\s+', ' ', regex=True).str.strip()
pdf = pdf.drop_duplicates(subset=['text'])
print('pandas', round(time.perf_counter() - t0, 2), 's', len(pdf))
# 预期: pandas 18.4 s  4210933

t0 = time.perf_counter()
ldf = (pl.scan_ndjson('raw.jsonl')          # lazy：谓词 / 投影下推
         .filter(pl.col('text').str.len_chars() > 50)
         .with_columns(pl.col('text').str.replace_all(r'\s+', ' ').str.strip_chars())
         .unique(subset=['text'])
         .collect())
print('polars', round(time.perf_counter() - t0, 2), 's', ldf.height)
# 预期: polars 2.1 s  4210933   ← 约 8–10x，峰值内存也显著更低

3.3 动手练习与自测（数据工程与 SQL）

1. 写一条 SQL，从样本表按「用户哈希」切成 train/valid/test，并附一条泄漏自检查询。判据：用 CASE WHEN user_id % 10 ... 或哈希取模按用户切；自检用 GROUP BY user_id HAVING COUNT(DISTINCT split) > 1 必须为空。
2. 解释 ROWS BETWEEN ... 与 RANGE BETWEEN ... 的区别，并说明默认帧是什么。参考答案：ROWS 按物理行，RANGE 按排序「值」；默认帧是 RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW，对并列值会一起纳入，是累计结果与直觉不符的常见原因。
3. 为什么随机抽样构造评测集通常不可取？判据：高频意图会淹没有效样本，分布偏斜；应按意图 / 长度 / 语言 / 难度分层各取 N 条。
4. 用 polars 写出「读 JSONL → 过滤 → 正则归一 → 精确去重」的 lazy 管线；说明为何比 pandas 快。判据：用 scan_ndjson + filter + with_columns + unique + collect；快在 Rust 多核并行 + 惰性谓词 / 投影下推 + Arrow 列式布局。
5. 精确去重、MinHash、embedding 聚类分别解决什么、代价如何？判据：精确抓逐字节重复（最便宜）；MinHash 抓高 Jaccard 近重复；embedding 抓语义重复但最贵。三者是递进的成本—收益关系，通常按此顺序叠加使用。

4. Git / Linux / Docker / CI：部署前的最后一公里

4.1 Git 工作流与 AI 项目的特殊要求

• 分支策略：main 保护 + 短生命周期特性分支 + PR 评审。AI 项目额外要注意：实验分支不要长期存在，实验参数进配置而不是散在代码里，否则三个月后没人能还原。
• 大文件与管理：模型权重、数据集绝不进 git。用 Git LFS（小规模）或对象存储 + 版本清单（推荐）。.gitignore 要提前写好 *.pt、*.safetensors、data/、.env、wandb/。
• 密钥泄露是最常见的严重事故：.env 必须在第一次 commit 前进 .gitignore。万一泄露，改密码不等于安全——必须轮换密钥并清理历史（git filter-repo / BFG）。
• Prompt 与配置也要进版本管理：Prompt 是代码的一部分。改 Prompt 要能 diff、能回滚、能追溯「哪次改动导致指标下降」。这是 AI 项目与传统项目最大的差异之一。

bash# 高频命令（AI 项目里最常用的那些）
git switch -c feat/retrieval-hnsw          # 建特性分支
git add -p                                  # 逐块暂存，避免把调试代码一起提交
git commit -m 'feat(index): 实现 HNSW 索引与召回率基准'
git rebase -i main                          # 整理提交历史（不要 rebase 已推送的公共分支）
git cherry-pick <sha>                       # 把某个修复搬到另一个分支
git stash push -m 'wip: 调参中的实验'        # 临时保存实验状态
git log --oneline --graph --all -20
git diff --stat HEAD~3                      # 看最近三次改了哪些文件

# 事故恢复
git reflog                                  # 「我丢了 commit」的救命稻草
git switch -c rescue <sha>                  # 从 reflog 找到的 sha 恢复

# 检查有没有不小心提交敏感文件
git log --all --full-history -- '*.env' '*.pem' '*credentials*'
git secrets --scan-history   # 或用 gitleaks

# 加 pre-commit 钩子（把 lint / 格式化 / 密钥扫描自动化）
cat > .pre-commit-config.yaml <<'YAML'
repos:
  - repo: https://github.com/astral-sh/ruff-pre-commit
    rev: v0.6.0
    hooks: [{ id: ruff }, { id: ruff-format }]
  - repo: https://github.com/gitleaks/gitleaks
    rev: v8.18.0
    hooks: [{ id: gitleaks }]
YAML
pre-commit install

bash# ① 定位「哪次提交把指标搞坏了」：git bisect 自动二分
git bisect start HEAD v1.2.0            # 坏的一端 HEAD，好的一端 v1.2.0
git bisect run ./scripts/eval_gate.sh   # 脚本退出码非 0 即判为「坏」，自动二分
# 100 个提交里藏一个回归，最多约 log2(100)≈7 次就定位到

# ② 并行开发多个实验分支，不要来回 stash：用 worktree
git worktree add ../orion-exp1 feat/lora-rank   # 另开目录检出分支

# ③ 误提交大文件后清理历史（谨慎：会重写历史，需团队同步强推）
git filter-repo --path data/ --invert-paths     # 或 BFG Repo-Cleaner

# ④ 强约定：提交信息可读，便于将来二分
#   <type>(<scope>): <subject>
#   例如  fix(rag): 修正 chunk 重叠导致重复召回

4.2 Linux 与 Shell：无 GUI 服务器上的排障

bash# ── 进程与端口（「服务起来了但访问不到」的标准排查顺序）──
ss -ltnp | grep 8000          # 端口到底有没有监听？监听在 0.0.0.0 还是 127.0.0.1？
ps aux | grep -i hamauls_orion        # 进程在不在？是不是启动后就退了
lsof -p <PID> | head -40      # 这个进程打开了哪些文件与端口
strace -p <PID> -f -e trace=network   # 卡住时看它在做什么系统调用

# ── 资源与显存 ──
df -h / free -h / uptime      # 磁盘 / 内存 / 负载
nvidia-smi                    # 显存是否被吃满、有没有别的进程占着卡
watch -n1 nvidia-smi          # 观察推理或训练是否真的在跑（利用率是否在跳）
nvidia-smi --query-compute-apps=pid,used_memory --format=csv   # 谁占了显存

# ── 日志（不要 cat 一个 10GB 的日志）──
tail -F app.log                      # 实时跟踪
grep -n 'ERROR\|Traceback' app.log | tail -50
journalctl -u hamauls-orion -f --since '10 min ago'
awk '{print $1}' access.log | sort | uniq -c | sort -rn | head   # 统计访问来源

# ── 传大文件（别用 scp 传几十 GB 模型）──
rsync -avhP --partial --inplace ./ckpt/ user@host:/data/ckpt/    # 支持断点续传
# ── 定时与并发控制 ──
nohup python train.py > train.log 2>&1 & echo $! > train.pid     # 后台跑并记 PID
flock -n /tmp/train.lock python train.py                          # 防止重复启动

bash# ── 判断瓶颈在哪：CPU / 内存 / 磁盘 / 网络，四选一 ──
vmstat 1 5            # r>核数→CPU 排队；b>0→阻塞；si/so 非 0→在换页（危险）
iostat -xz 1          # %util 接近 100 且 await 高 → 磁盘饱和
sar -n DEV 1 5        # 网卡是否打满（对比链路带宽）
pidstat -p <PID> 1    # 单进程的 CPU / IO 分解
# 经验判据：r 长期 > 2×核数 = CPU 瓶颈；
#          si/so 持续非 0 = 内存不足在 swap（延迟灾难）

# ── 服务被 OOM Killer 杀掉后怎么确认 ──
dmesg -T | grep -i 'killed process'      # 会有 Out of memory: Killed process <pid>
journalctl -k --since '1 hour ago' | grep -i oom

# ── 提高文件描述符上限（高并发服务必调）──
ulimit -n                                # 默认常是 1024，千级并发会报 too many open files
cat /proc/<PID>/limits | grep 'open files'
# 永久修改：/etc/security/limits.conf 设 nofile 65535，或 systemd 的 LimitNOFILE=

4.3 Docker 与 CI：一条命令跑起来

dockerfile# 多阶段构建的 Dockerfile（分层缓存 + 非 root 运行 + 依赖与代码分离）
FROM python:3.12-slim AS builder
ENV PIP_NO_CACHE_DIR=1
WORKDIR /app
# ① 先装依赖（变更少）再拷代码（变更多）—— 充分利用镜像层缓存
COPY pyproject.toml uv.lock ./
RUN pip install uv && uv sync --frozen --no-dev

FROM python:3.12-slim AS runtime
ENV PYTHONUNBUFFERED=1 PYTHONDONTWRITEBYTECODE=1
WORKDIR /app
COPY --from=builder /app/.venv /app/.venv
ENV PATH='/app/.venv/bin:$PATH'
COPY src/ ./src/
COPY configs/ ./configs/
# ② 非 root 运行（安全基线，很多合规检查会卡这一条）
RUN useradd -m app && chown -R app:app /app
USER app
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=3s \
  CMD python -c 'import urllib.request,sys; sys.exit(0 if urllib.request.urlopen("http://localhost:8000/health").status==200 else 1)'
CMD ['uvicorn', 'hamauls_orion.serving.api:app', '--host', '0.0.0.0', '--port', '8000']

yaml# .gitlab-ci.yml —— M1 的验收条件之一：CI 全绿
stages: [test]

test:
  stage: test
  image: python:3.12-slim
  before_script:
    - pip install uv                       # GitLab Runner 里装 uv
  script:
    - uv sync --frozen            # 严格按锁文件安装，保证可复现
    - uv run ruff check src tests        # 静态检查
    - uv run ruff format --check src tests
    - uv run mypy src                    # 类型检查
    - uv run pytest -q -m 'not slow' --cov=src --cov-fail-under=70
    # 覆盖率门槛：不是为了数字好看，而是防止「新代码完全没测」

• 分层缓存的收益是巨大的：把「装依赖」放在「拷代码」之前，代码改动时不需要重装依赖，镜像构建时间从几分钟降到几秒。这是 Docker 优化里性价比最高的一条。
• 多阶段构建：builder 阶段装编译工具与依赖，runtime 阶段只拷贝产物。镜像体积能小 5–10 倍，也减少了攻击面。
• GPU 容器：需要装 NVIDIA Container Toolkit，运行时用 --gpus all，基础镜像换成 nvidia/cuda:...-runtime-ubuntu22.04。注意要选 runtime 而不是 devel（后者体积大得多，除非你要编译 CUDA 代码）。
• compose 一键起依赖：向量库、Redis、Postgres 用 compose 起在本地，让「docker compose up 就能开发」成为团队的默认体验。

yaml# docker-compose.yaml —— 一键起本地依赖，让「clone 即开发」成立
services:
  api:
    build: { context: ., dockerfile: docker/Dockerfile }
    ports: ['8000:8000']
    env_file: [.env]
    depends_on:
      postgres: { condition: service_healthy }   # 等依赖真的就绪，而非「起来了」
    volumes: ['./data:/app/data:ro']             # 代码进镜像，数据挂卷
  postgres:
    image: pgvector/pgvector:pg16
    environment:
      POSTGRES_PASSWORD: dev
    healthcheck:
      test: ['CMD-SHELL', 'pg_isready -U postgres']
      interval: 5s
      retries: 10
  redis:
    image: redis:7-alpine
    command: ['redis-server', '--maxmemory', '512mb', '--maxmemory-policy', 'allkeys-lru']

4.4 动手练习与自测（Git / Linux / Docker / CI）

1. 服务「起来了但外部访问不到」，写出你的排查清单与判据。参考答案：ss -ltnp | grep 看是否监听及绑定地址（0.0.0.0 vs 127.0.0.1）→ ps aux 看进程是否退出 → free -h / df -h 看内存磁盘 → tail -F / journalctl -u 看日志 → 再查防火墙与依赖连通性。
2. 用 git bisect 在 128 个提交里定位一个回归，最多需要多少次验证？判据：约 log2(128)=7 次；配合 git bisect run <脚本> 可自动二分（脚本退出码非 0 判为「坏」）。
3. 把「装依赖」放在「拷代码」之前为什么能显著加快构建？参考答案：Docker 按层缓存，代码改动只影响后面的层；若顺序反了，任何代码改动都会使依赖层缓存失效、重装全部依赖，构建从几秒变回几分钟。
4. GPU 容器要做什么额外配置？基础镜像怎么选？判据：宿主装 NVIDIA Container Toolkit，运行时 --gpus all；基础镜像选 nvidia/cuda:...-runtime（约 2–3 GB）而非 -devel（约 6–9 GB），后者仅编译 CUDA 时才需要。
5. 进程「活着但不可用」有哪些典型成因？判据：监听地址错（127.0.0.1）、事件循环被同步调用阻塞、显存被占满、磁盘写满、依赖未就绪；对策是健康检查真正打业务路径，并区分 liveness 与 readiness。

项目里程碑

阶段练习项目

常见误区

• 把 notebook 当工程：没有类型标注、没有日志、没有测试，变量随手改名，三个月后自己都读不懂。
• 在 async 代码里调用同步阻塞函数（requests / time.sleep / 某些 SDK），以为「写了 async 就并发了」，实际整个事件循环被卡死。
• 用 `except Exception` 一把抓，把真正的 bug 也吞掉了，问题在生产环境才暴露。
• 不设并发上限，把对端限流打满后被封，或者把本地内存打爆。
• 把所有密钥、IP、模型名硬编码进代码，换环境要改代码而不是改配置；或者把 `.env` 提交进了仓库。
• 跳过数据质量直接调模型，然后花几周怀疑是模型不行。
• 日志用 f-string 拼接，在内层循环里付出了大量无谓的格式化开销。
• 只学 Python 完全不懂前端与 Docker，导致永远无法独立交付一个完整产品。
• 把模型权重与数据集塞进 git，仓库膨胀到无法克隆。
• 没有配置分层，实验参数散落在脚本里，无法复现任何一次实验。

面试高频问题速答

Python 的 GIL 是什么，对 AI 场景有什么影响？

GIL（全局解释器锁）让同一进程内同一时刻只有一个线程执行 Python 字节码，所以多线程无法并行做 CPU 计算——但它会在 I/O 等待（网络、磁盘）时释放，也会在 numpy / torch 这类 C 扩展执行时释放（它们在 C 层主动释放 GIL）。结论：模型 API 调用、文件与网络 I/O 用 asyncio 或多线程都有效；而 tokenize、纯 Python 数值计算这类 CPU 密集任务必须用多进程（ProcessPoolExecutor）或换成向量化 / C 扩展。注意：CPython 3.13+ 的 free-threaded 构建（PEP 703）正在逐步解除这个限制，但生态适配仍在进行中，生产环境暂不宜依赖。

asyncio 和线程池怎么选？

看依赖的接口形态。① 全链路可 async（aiohttp、asyncpg、MCP 的 async 客户端）→ 用 asyncio，单机可支撑上千并发连接，内存开销小（每个协程几 KB，而每个线程约 1 MB 栈）。② 必须调用只有同步接口的库（部分 SDK、某些数据库驱动）→ 用 asyncio.to_thread 或线程池包一层，把阻塞调用挪出事件循环。③ CPU 密集（tokenize、特征计算）→ ProcessPoolExecutor 或多进程。实际项目里常常三者混用：asyncio 做骨架、to_thread 包同步库、进程池做重计算。

怎么保证 LLM 应用的输出可控？

三层防御：① 约束生成——用结构化输出（JSON Schema / function calling / grammar 约束解码）让模型只能在合法空间里输出；② 边界校验——在系统边界用 Pydantic 严格校验，校验失败走降级或有限重试，而不是崩溃；如果模型输出要渲染成 HTML，还必须走安全的 markdown 渲染防 XSS；③ 回归门禁——用 Evals（黄金数据集 + 结构化断言 + LLM-as-Judge）在 CI 里做门禁，指标下降就阻断合并，而不是靠人工目测。这三层缺任何一层，都会在真实流量下暴露问题。

一次「效果变差」的线上问题，你的排查顺序？

先对齐变量：数据版本、Prompt 版本、模型版本、采样参数有没有变——绝大多数问题在这里，所以第一步永远是「有没有可回溯的实验记录」。再查链路：检索是否召回变差、上下文是否被截断、工具调用是否失败、是否命中了缓存返回了旧结果。再看流量分布：是否出现了新的输入分布（新用户群体、新语言、超长输入）。最后才怀疑模型本身。这个顺序的价值在于：它把「最可能的原因」排在最前面，避免在低概率方向上浪费几小时。

Docker 在 AI 项目里的价值是什么？有哪些坑？

价值：① 解决「我这能跑你那儿不能跑」——固定 CUDA / Python / 依赖版本；② 镜像分层缓存让构建与部署快；③ 配合 compose 一键起本地依赖（向量库、缓存、数据库）；④ 让实验环境与生产环境尽量一致。坑：① GPU 场景需要 NVIDIA Container Toolkit 与 `--gpus`，且必须选 `runtime` 而非 `devel` 基础镜像控制体积；② 镜像里不要放模型权重（体积巨大且更新频繁），应该运行时挂载卷或从对象存储拉取；③ 容器内 `localhost` 不是宿主机的 localhost，网络模型要理清；④ 非 root 运行时挂载卷的权限问题很常见；⑤ 构建机与运行机的 CPU 指令集差异（AVX 支持）可能导致「镜像能跑但 numpy 报 illegal instruction」。

为什么项目要把源码放在 src/ 目录下？

因为 Python 的模块查找会把「当前工作目录」加入 sys.path。如果源码直接在仓库根目录，那么 `python scripts/run.py` 时 `import hamauls_orion` 会在本地目录直接命中，即使项目的打包配置是错的也能跑通——于是问题被推迟到安装之后才暴露（比如漏了包数据、`packages` 配置不对、循环依赖）。`src/` 布局强制你以「已安装的包」方式测试（通常配合 `pip install -e .`），能提前发现打包问题。这在需要发布内部包或部署到容器的项目里是必须的。

学习资源