企业官网建设流程全解析

这两年 AI 相关的岗位被讨论得非常多，但真正落到"怎么从零开始、按什么顺序学、学到什么程度算能上手"的资料其实很零散。很多同学卡在两个极端之间：一边是铺天盖地的算法课和论文推导，学完不知道能做什么；另一边是各种"三行代码调用大模型"的速成贴，跑通了 demo，一到交付就发现鉴权、额度、检索质量、成本控制全都不会做。

这篇文章想解决的就是中间这段路。标题里的"最快""从零到 SaaS"我不打算写成口号，也不承诺任何收入数字——薪资是结果，不是路径。本文只讨论一件可验证的事：一个普通后端或前端开发者，需要补齐哪些具体能力、按什么顺序练、用一个什么样的项目把能力串起来，最后能独立交付一个带鉴权、带额度控制、带检索增强、能容器化部署的 AI 应用。

文章定位偏实战：有明确的技术栈，有可直接复制的代码，有从零跑通再到上线的完整流程，也有踩坑排查清单。适合正在转型的后端/前端开发者、想做一些 AI 应用练手的学生，以及需要给团队做技术选型的工程同学。读完之后，你至少能自己搭起一个结构清晰、可扩展、能上线的文档问答类 AI 应用，并且知道下一步该往哪里深入。

1. 先把概念边界划清楚：AI 工程师到底在做什么

1.1 三类容易混淆的岗位

开始学之前，最好先搞清楚几个常被混用的岗位名称，否则很容易学错方向。

第一类是算法研究员 / 机器学习工程师，主要工作是数据清洗、模型结构设计、训练调优、评测指标提升，产出物通常是模型权重或训练框架，日常打交道的是 PyTorch、分布式训练、数据流水线。这条路的门槛在数学和系统工程，周期长，岗位数量相对少。

第二类是AI 应用工程师（AI Engineer），也就是这篇文章主要讨论的方向。核心工作是把已经具备能力的模型，接进一个真实业务系统里，让它稳定、可控、可计费地对外提供服务。日常打交道的是 API 调用、上下文设计、检索增强、服务框架、部署和监控，产出物是一个能跑的产品。

第三类是数据工程师 / 平台工程师，负责数据接入、向量化管道、特征存储、模型服务化基础设施，偏向底座建设。

三者不是高低关系，而是职责边界不同。2026 年市场需求增长最快的，其实是第二类。原因很直接：基础模型能力已经足够强，绝大多数业务不需要自己训练模型，缺的是"把模型能力可靠地交付出去"的人。

1.2 AI 应用工程师的一天在做什么

很多同学以为这个岗位就是写 Prompt，实际上 Prompt 只占很小一部分。一个典型的迭代周期大概是这样：

需求进来之后，先判断任务是否适合用模型解决，还是用规则、检索、传统算法更划算。确定要用模型之后，进入模型选型和上下文设计，考虑用哪个模型、上下文窗口多大、是否需要拆成多步、输出要不要约束成结构化 JSON。然后处理数据侧，检索场景要把文档切块、建立索引、调召回质量。接着是服务侧封装，把调用逻辑包成接口，加上超时、重试、降级、额度、日志。最后是评估和上线，用一批回归用例验证改动没有把事情变坏，观察线上延迟、错误率、单次成本。

可以看到，除了第二步稍微"AI"一点，其余全是常规工程能力的延伸。这也是为什么后端转 AI 应用工程师的路径通常比零基础转算法岗短得多——你已经有的接口设计、数据库、容器化经验，全都直接可用。

1.3 为什么现在强调"工程"而不是"训练"

一个常见误区是把"学 AI"等价于"学模型训练"。在业务场景里，微调模型往往是最后手段，而不是第一选择。原因是成本结构：一次微调要考虑数据标注、训练资源、效果回归、上线灰度，而通过更好的上下文组织、更准确的检索结果、更严格的输出约束，往往能解决 70% 以上的问题，改动成本只有前者的零头。

所以本文的路线会把重心放在上下文工程、检索增强、服务化与成本控制上。这些能力不依赖某个特定模型，换模型也不会失效，是真正的可迁移技能。

2. 能力地图：从零到能交付的学习顺序

这一节给出的是顺序，而不是清单。顺序错了，最常见的结果是学了一堆碎片，但拼不出一个东西。

2.1 第一阶段：能写服务，而不是写脚本

这是很多人忽略的一步。会用 Python 写脚本，和会用 Python 写一个能对外提供服务的应用，中间隔着一层。需要掌握的具体能力包括：用框架（如 FastAPI）定义路由和请求模型、处理请求校验、理解同步与异步的差别、知道什么是连接池、会用环境变量管理配置、能读懂调用栈里抛出的异常。

判断标准很简单：让你从零写一个带三个接口、有参数校验、有错误码的小服务，能不能在半小时内跑起来。

2.2 第二阶段：模型调用与上下文工程

这一层的核心不是"会不会调用接口"，而是"能不能控制输出"。需要掌握的内容包括：模型调用协议的统一封装（同一套代码能否切换不同服务端点）、消息角色（system / user / assistant）的作用与顺序影响、上下文窗口和 token 预算的估算、如何让模型返回稳定可解析的结构化数据、超时与重试的处理方式、流式返回的实现。

判断标准：给定一个业务需求，你能写出稳定的系统提示，并让模型在 95% 以上的情况下返回符合 schema 的结果。

2.3 第三阶段：检索增强与数据管道

这是让模型"知道你的东西"的关键。需要掌握：文档解析与切块策略（切多大、重叠多少、按什么边界切）、索引与召回（先用简单方案跑通，再换向量库）、召回结果的重排与拼装、引用与溯源如何呈现给用户。

判断标准：给一份 20 页的产品手册，你的系统能回答手册里的细节问题，并且指出来源段落。

2.4 第四阶段：产品化与运维

这一步是把 demo 变成能收费的东西。需要掌握：鉴权方式、按用户计费或配额控制、调用日志与成本统计、容器化打包、反向代理与 TLS、错误码与降级策略、基础监控指标。

判断标准：你能把整个应用打包成一个镜像，用一条命令在服务器上启动，并且能看到每个用户的用量。

下面这张表把四个阶段和常见误区放在一起，方便自查：

3. 环境准备与项目结构

3.1 运行环境说明

本文示例使用的是一套通用环境，不作为唯一标准。Python 建议 3.10 及以上版本（用到较多类型标注语法），操作系统 Linux 或 macOS 均可，Windows 建议在 WSL 下操作以免遇到路径和编码问题。容器相关部分需要 Docker 与 Docker Compose。

模型服务方面，本文示例通过 OpenAI 兼容协议访问模型。这个协议目前被广泛支持：无论是云端厂商提供的兼容端点，还是本地部署的开源模型服务，大多都遵循同一套请求格式。因此示例代码里的base_url和api_key都是可配置项，你把它指向自己实际使用的服务即可，代码逻辑不需要改。

需要特别说明：具体依赖版本请根据你的项目实际情况调整。本文示例重点演示配置思路和代码结构，不绑定某一个具体版本号，避免因为版本迭代导致示例失效。

3.2 依赖清单

# 文件路径：requirements.txt fastapi>=0.110 uvicorn[standard]>=0.27 openai>=1.30 pydantic>=2.6 python-dotenv>=1.0 python-multipart>=0.0.9

这里每一项都有明确用途：fastapi提供路由和请求模型，uvicorn是 ASGI 服务器，openai是官方 SDK（同时兼容各种遵循同一协议的端点），pydantic负责请求与响应的数据校验，python-dotenv用于读取.env文件，python-multipart支持文件上传表单。

之所以没有引入向量数据库、重排模型、Agent 框架，是因为第一步的目标是跑通闭环。过早引入重型依赖，会让你在出问题时不知道是哪一层坏了。

3.3 环境变量与密钥管理

密钥管理是新手最容易踩的坑。绝对不要在代码里硬编码api_key，也不要提交到代码仓库。推荐做法是本地用.env文件，服务器上用环境变量注入。

# 文件路径：.env（不要提交到版本库） MODEL_API_KEY=your_key_here MODEL_BASE_URL=https://your-compatible-endpoint/v1 MODEL_NAME=your-model-name DB_PATH=data/app.db DAILY_QUOTA=200 CHUNK_SIZE=500 TOP_K=4

同时务必添加.gitignore：

# 文件路径：.gitignore .env data/ __pycache__/ *.pyc .venv/

在生产环境，密钥应通过容器编排平台的密钥管理能力注入，遵循最小权限原则：只授予模型调用所需权限，不要用管理员密钥跑业务。如果密钥发生泄漏，第一时间轮换，而不是指望没人发现。

3.4 项目目录结构

一个清晰的结构能省下大量后期重构时间。本文示例采用如下布局：

ai-saas-demo/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口，注册路由与启动钩子 │ ├── config.py # 配置读取 │ ├── db.py # SQLite 数据访问 │ ├── llm.py # 模型调用封装 │ ├── rag.py # 切块与检索 │ └── routers/ │ ├── __init__.py │ ├── chat.py # 对话接口 │ └── admin.py # 文档上传与用量查询 ├── data/ │ └── app.db ├── web/ │ └── index.html # 最小前端页面 ├── requirements.txt ├── Dockerfile ├── docker-compose.yml └── .env

按职责拆分文件而不是全部塞进main.py，是为了让"模型调用"和"业务逻辑"解耦。后面你要换模型、换检索方案、加缓存，都只需要动一个文件。

4. 核心技术点拆解

4.1 模型调用的统一封装

把调用逻辑集中到一个文件，是最简单也最有效的工程习惯。好处有三个：换服务端点只改一处；超时、重试、日志可以统一加；测试时可以方便地打桩替换。

# 文件路径：app/config.py import os from dataclasses import dataclass from dotenv import load_dotenv load_dotenv() @dataclass(frozen=True) class Settings: api_key: str = os.getenv("MODEL_API_KEY", "") base_url: str = os.getenv("MODEL_BASE_URL", "") model: str = os.getenv("MODEL_NAME", "") db_path: str = os.getenv("DB_PATH", "data/app.db") daily_quota: int = int(os.getenv("DAILY_QUOTA", "200")) chunk_size: int = int(os.getenv("CHUNK_SIZE", "500")) top_k: int = int(os.getenv("TOP_K", "4")) settings = Settings()

这里用dataclass(frozen=True)而不是普通类，是为了让配置在进程内只读，避免运行中被误改。load_dotenv()放在模块顶部，保证任何导入settings的地方都能拿到值。

# 文件路径：app/llm.py from openai import OpenAI from app.config import settings _client = OpenAI( api_key=settings.api_key, base_url=settings.base_url or None, timeout=60.0, max_retries=2, ) def chat_completion(messages, temperature=0.2, max_tokens=800, stream=False): """统一的对话调用入口，便于后续替换模型或加缓存。""" return _client.chat.completions.create( model=settings.model, messages=messages, temperature=temperature, max_tokens=max_tokens, stream=stream, )

关键参数说明：timeout一定要设，模型调用属于典型的长尾请求，不设超时会让整个线程被拖住；max_retries交给 SDK 处理瞬时故障，比自己在业务层写重试更可靠；temperature在问答类场景建议调低（0.1～0.3），减少随机性带来的答案漂移。

常见误区是把base_url写死。一旦服务端点变更或者要做多活，就得改代码重新发版。养成从配置读的习惯，后期能省很多事。

4.2 上下文工程：系统提示怎么写才稳定

很多人写 Prompt 的毛病是"想到什么写什么"。稳定的系统提示通常包含五个部分：角色定义、任务边界、输入说明、输出格式约束、拒答与兜底规则。

一个可以直接用的模板结构如下：

你是【产品名】的知识库助手。 任务：仅根据下面提供的【参考资料】回答用户问题。 约束： 1. 参考资料中没有的信息，直接回答"资料中未提及"，不要自行推测。 2. 回答控制在 200 字以内，使用中文。 3. 如果引用了具体段落，在答案末尾用 [编号] 标注来源。 4. 不要输出与问题无关的寒暄。 【参考资料】 {context} 【用户问题】 {question}

这份模板里最关键的是第 1 条。没有这条，模型会非常自信地编造内容，这是线上事故的高发点。第 3 条让答案可溯源，出问题时你能快速判断是检索错了还是模型胡说了。

还有一点容易被忽略：参考资料和用户问题要放在不同的分隔块里。如果直接把用户输入拼在指令后面，用户完全可以用"忽略上述要求"这类输入覆盖你的系统提示。把用户问题用明确标记包起来，能显著降低这种风险。

4.3 结构化输出：让模型返回可解析的数据

只要你的下游需要程序处理模型输出，就不要指望自由文本。常见做法是明确要求返回 JSON，并给出字段示例。

# 文件路径：app/llm.py（续） import json SCHEMA_PROMPT = """你是一个信息抽取器。请从用户输入中抽取字段，严格返回 JSON，不要输出任何其他内容。 JSON 结构： {"intent": "字符串，取值只能是 chat 或 search", "keywords": ["字符串数组"], "need_source": true 或 false} 如果无法判断，intent 填 chat，keywords 填空数组。""" def extract_fields(text: str) -> dict: resp = chat_completion( messages=[ {"role": "system", "content": SCHEMA_PROMPT}, {"role": "user", "content": text}, ], temperature=0.0, max_tokens=300, ) raw = resp.choices[0].message.content.strip() # 兼容模型偶尔包裹代码块的情况 if raw.startswith("```"): raw = raw.strip("`") if raw.lower().startswith("json"): raw = raw[4:] try: return json.loads(raw) except json.JSONDecodeError: return {"intent": "chat", "keywords": [], "need_source": False}

这段代码有两个工程细节值得注意。第一是temperature=0.0，抽取类任务不需要创造性。第二是异常兜底：即使做了约束，模型仍有概率返回带 Markdown 代码块包裹的内容，必须做清洗，并且在解析失败时返回一个默认值，而不是让整个请求抛 500。

如果你的服务端点支持结构化输出或函数调用能力，优先用原生能力，比纯文本约束更稳。但兜底逻辑依然要留，因为不同端点的支持程度不一致。

4.4 RAG 的最小闭环

RAG（Retrieval-Augmented Generation，检索增强生成）听起来复杂，最小闭环只有四步：切块、建索引、召回、拼装。

切块的关键参数是块大小和重叠量。块太小，语义不完整，检索出来的片段答不了问题；块太大，噪声多，挤占上下文窗口。中文文档的经验值在 300～800 字之间，重叠 50～100 字，并且尽量按段落或标题边界切，而不是机械地每 N 个字切一刀。

索引和召回是方案差异最大的地方。生产环境通常用向量检索加关键词检索的混合方案，向量库可选范围很广。但为了让你第一步能跑通、能调试、能看懂每一分是怎么来的，本文示例先用一个纯 Python 的轻量打分方案：基于字符片段重叠和词频统计。它的召回质量不如向量检索，但零依赖、可解释、便于建立基线。等你把闭环跑通、有了评估集，再替换成向量方案，就能清楚知道"换方案到底带来了多少提升"。

# 文件路径：app/rag.py import re from collections import Counter from app.config import settings _CN = re.compile(r"[\u4e00-\u9fa5]") _EN = re.compile(r"[a-zA-Z0-9]+") def split_text(text: str, size: int = None, overlap: int = 80): """按段落优先切块，段落过长再按长度硬切。""" size = size or settings.chunk_size paragraphs = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] chunks, buf = [], "" for p in paragraphs: if len(buf) + len(p) + 1 <= size: buf = f"{buf}\n{p}".strip() else: if buf: chunks.append(buf) while len(p) > size: chunks.append(p[:size]) p = p[size - overlap:] buf = p if buf: chunks.append(buf) return chunks def _tokens(text: str): """中文取二字片段，英文取整词，兼顾中文检索效果。""" tokens = [] zh = "".join(_CN.findall(text)) for i in range(len(zh) - 1): tokens.append(zh[i:i + 2]) tokens.extend(w.lower() for w in _EN.findall(text)) return tokens def score(query: str, chunk: str) -> float: q = Counter(_tokens(query)) c = Counter(_tokens(chunk)) if not q or not c: return 0.0 hit = sum(min(v, c.get(k, 0)) for k, v in q.items()) return hit / (sum(q.values()) ** 0.5 * sum(c.values()) ** 0.5 + 1e-6) def retrieve(query: str, chunks, top_k: int = None): top_k = top_k or settings.top_k ranked = sorted(((score(query, c), c) for c in chunks), key=lambda x: x[0], reverse=True) return [c for s, c in ranked[:top_k] if s > 0] def build_context(chunks) -> str: """把召回片段编号拼装，方便模型引用来源。""" return "\n\n".join(f"[{i + 1}] {c}" for i, c in enumerate(chunks))

这里的score函数用的是归一化重叠度，思路和余弦相似度一致，只是把向量的"维度"换成了词片段。它的优点是完全可解释：你能打印出命中了哪些片段，判断是切块问题还是查询表述问题。

需要注意retrieve里的if s > 0。这是个容易被忽略的细节：如果所有片段得分都是 0，说明知识库里根本没有相关内容，此时应该让模型回答"资料中未提及"，而不是硬塞几段无关内容进去，那样只会诱导模型编造。

4.5 额度与成本控制：第一天就要做

这是个人项目和商用产品最大的分界线。模型调用是按量付费的，一旦对外开放又没有限制，最好的情况是账单失控，最坏的情况是密钥被滥用。

最小可行的方案是按天配额加调用日志。为什么不按调用次数总限额？因为日配额能更好地控制峰值和突发，也更容易向用户解释。

数据模型上，两张表就够：用户表存密钥和套餐，用量表存每日计数。

# 文件路径：app/db.py import os import sqlite3 from datetime import date def _conn(): os.makedirs(os.path.dirname(os.getenv("DB_PATH", "data/app.db")), exist_ok=True) conn = sqlite3.connect(os.getenv("DB_PATH", "data/app.db")) conn.execute("PRAGMA journal_mode=WAL") return conn def init_db(): conn = _conn() try: conn.execute( """CREATE TABLE IF NOT EXISTS users ( api_key TEXT PRIMARY KEY, plan TEXT NOT NULL DEFAULT 'free', created_at TEXT NOT NULL )""" ) conn.execute( """CREATE TABLE IF NOT EXISTS usage_daily ( api_key TEXT NOT NULL, day TEXT NOT NULL, used INTEGER NOT NULL DEFAULT 0, PRIMARY KEY (api_key, day) )""" ) conn.commit() finally: conn.close() def ensure_user(api_key: str, plan: str = "free"): conn = _conn() try: conn.execute( "INSERT OR IGNORE INTO users(api_key, plan, created_at) VALUES (?, ?, ?)", (api_key, plan, date.today().isoformat()), ) conn.commit() finally: conn.close() def get_quota(api_key: str) -> int: conn = _conn() try: row = conn.execute( "SELECT used FROM usage_daily WHERE api_key=? AND day=?", (api_key, date.today().isoformat()), ).fetchone() return row[0] if row else 0 finally: conn.close() def incr_quota(api_key: str) -> int: """先占额度再调用模型，避免并发下超发。""" conn = _conn() try: conn.execute( """INSERT INTO usage_daily(api_key, day, used) VALUES (?, ?, 1) ON CONFLICT(api_key, day) DO UPDATE SET used = used + 1""", (api_key, date.today().isoformat()), ) conn.commit() row = conn.execute( "SELECT used FROM usage_daily WHERE api_key=? AND day=?", (api_key, date.today().isoformat()), ).fetchone() return row[0] finally: conn.close()

incr_quota里用INSERT ... ON CONFLICT DO UPDATE是关键。如果先查询再更新，两个并发请求可能都读到"未超限"，然后都去调用模型，配额就超发了。用一条原子语句把"检查并占用"合并，是这类限流逻辑的标准写法。

SQLite 在这里够用，是因为它的写入是串行的，量级不大时反而比引入独立数据库更简单。如果用户量上来了，建议迁移到 PostgreSQL，并仿照上面的思路用行锁或原子更新保持一致性。

5. 完整实战：搭一个能上线的文档问答 SaaS

前面拆解了各个模块，这一节把它们组装起来。

5.1 需求与功能拆分

目标产品是这样一个小服务：用户拿到一个 API Key，通过接口上传文档，然后基于文档提问，每天有免费额度；超出额度返回 429；所有回答尽量基于文档内容并给出来源。

功能拆成四块：文档上传与切块入库、检索与问答、鉴权与配额、静态页面。

这里刻意没有做用户注册、在线支付。理由是把范围控制住：注册和支付属于标准 Web 能力，可以后补；而"检索质量"和"配额正确性"才是这个产品的核心难点，先做核心，再做外围，是减少返工的基本原则。

5.2 文档存储设计

单机场景下不需要向量库，用内存加文件就够了。为了让重启后文档不丢，把切块结果按会话保存成 JSON 文件。

# 文件路径：app/store.py import json import os import threading _LOCK = threading.Lock() _DATA_DIR = "data/docs" def _path(doc_id: str) -> str: return os.path.join(_DATA_DIR, f"{doc_id}.json") def save_doc(doc_id: str, chunks) -> None: os.makedirs(_DATA_DIR, exist_ok=True) with _LOCK: with open(_path(doc_id), "w", encoding="utf-8") as f: json.dump({"chunks": chunks}, f, ensure_ascii=False) def load_doc(doc_id: str): p = _path(doc_id) if not os.path.exists(p): return [] with open(p, encoding="utf-8") as f: return json.load(f).get("chunks", [])

加锁是因为 FastAPI 默认在线程池里执行同步函数，多个请求同时写同一个文件会出问题。虽然这个小项目并发很低，但养成"共享资源要加锁"的习惯，迁移到生产时能避免很多诡异问题。

5.3 对话接口实现

先写鉴权依赖，再写问答接口。

# 文件路径：app/security.py from fastapi import Header, HTTPException from app import db from app.config import settings def require_key(authorization: str = Header(default="")): """从 Authorization: Bearer <key> 中解析并校验配额。""" if not authorization.startswith("Bearer "): raise HTTPException(status_code=401, detail="缺少有效的 Authorization 头") key = authorization[7:].strip() if not key: raise HTTPException(status_code=401, detail="API Key 为空") db.ensure_user(key) used = db.get_quota(key) if used >= settings.daily_quota: raise HTTPException(status_code=429, detail="今日额度已用完，请明天再试") new_used = db.incr_quota(key) return {"api_key": key, "used": new_used}

注意这里把"额度占用"放在依赖里完成，而不是在业务函数里。好处是接口签名干净，而且只要你把这个依赖挂上，就不存在某个接口忘记检查的情况。

下面是对外问答接口，包含流式返回。

# 文件路径：app/routers/chat.py from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from app import store from app.llm import chat_completion from app.rag import retrieve, build_context from app.security import require_key router = APIRouter() class ChatRequest(BaseModel): doc_id: str = Field(..., description="文档标识") question: str = Field(..., min_length=1, max_length=2000) SYSTEM_TMPL = """你是知识库助手。请仅根据【参考资料】回答，资料中没有的信息回答"资料中未提及"。 回答控制在 200 字以内，并在引用处标注 [编号]。 【参考资料】 {context} """ @router.post("/api/chat") def chat(req: ChatRequest, auth: dict = Depends(require_key)): chunks = store.load_doc(req.doc_id) if not chunks: raise HTTPException(status_code=404, detail="文档不存在或尚未解析完成") hits = retrieve(req.question, chunks) context = build_context(hits) if hits else "（无相关内容）" messages = [ {"role": "system", "content": SYSTEM_TMPL.format(context=context)}, {"role": "user", "content": req.question}, ] def gen(): try: stream = chat_completion(messages, temperature=0.2, stream=True) for event in stream: if not event.choices: continue delta = event.choices[0].delta piece = getattr(delta, "content", None) if piece: yield piece except Exception as exc: # 保证异常不会让连接悬空 yield f"\n[生成中断：{type(exc).__name__}]" return StreamingResponse(gen(), media_type="text/plain; charset=utf-8")

几个实现要点值得说明。

第一，getattr(delta, "content", None)是必要的防御。流式返回中第一个事件通常是角色信息，content为None，直接拼接会抛TypeError，这是新手接流式接口最容易遇到的报错。

第二，try/except包在生成器内部而不是接口外层。因为StreamingResponse的生成器是在响应已经开始发送之后才执行的，此时再抛异常，HTTP 状态码已经发出去了，客户端只会看到连接中断，拿不到任何有用信息。改成在流里写一段提示文本，用户体验好得多。

第三，接口返回纯文本流而不是 SSE 格式。SSE 更规范，但前端解析要处理data:前缀和事件边界，对于这个最小页面来说反而增加复杂度。等你要做多事件类型（比如同时推送来源和正文）时再换 SSE。

文档上传接口放在管理路由里，同时加一个简单的域名白名单校验，避免被人当成公共存储用。

# 文件路径：app/routers/admin.py import hashlib from fastapi import APIRouter, Depends, HTTPException, UploadFile, File from app import store, db from app.rag import split_text from app.security import require_key router = APIRouter() MAX_BYTES = 2 * 1024 * 1024 @router.post("/api/docs") async def upload(file: UploadFile = File(...), auth: dict = Depends(require_key)): raw = await file.read() if len(raw) > MAX_BYTES: raise HTTPException(status_code=413, detail="文件超过 2MB 限制") try: text = raw.decode("utf-8") except UnicodeDecodeError: raise HTTPException(status_code=400, detail="请上传 UTF-8 编码的文本文件") chunks = split_text(text) if not chunks: raise HTTPException(status_code=400, detail="文档内容为空") doc_id = hashlib.sha256(raw).hexdigest()[:16] store.save_doc(doc_id, chunks) return {"doc_id": doc_id, "chunks": len(chunks)} @router.get("/api/usage") def usage(auth: dict = Depends(require_key)): return { "api_key_prefix": auth["api_key"][:6] + "***", "used_today": db.get_quota(auth["api_key"]), }

MAX_BYTES这个限制不是可选项。文件上传接口如果没有大小限制，一个恶意请求就能把内存打满。同样，decode("utf-8")失败要捕获并返回 400，而不是让它变成 500 错误。

5.4 应用入口与前端页面

# 文件路径：app/main.py from fastapi import FastAPI from fastapi.responses import FileResponse from app.db import init_db from app.routers import chat, admin app = FastAPI(title="AI 文档问答示例") @app.on_event("startup") def _startup(): init_db() app.include_router(chat.router) app.include_router(admin.router) @app.get("/") def index(): return FileResponse("web/index.html")

前端用一个 HTML 页面就够，重点是演示流式读取。

<!-- 文件路径：web/index.html --> <!DOCTYPE html> <html lang="zh-CN"> <head> <meta charset="utf-8" /> <meta name="viewport" content="width=device-width, initial-scale=1" /> <title>文档问答示例</title> <style> body { max-width: 760px; margin: 40px auto; font-family: system-ui, sans-serif; line-height: 1.7; } input, button, textarea { font-size: 15px; padding: 8px; } textarea { width: 100%; height: 80px; } #answer { white-space: pre-wrap; background: #f6f8fa; padding: 12px; border-radius: 8px; margin-top: 12px; } </style> </head> <body> <h3>文档问答示例</h3> <p><input id="key" placeholder="API Key" style="width:100%" /></p> <p><input id="doc" placeholder="doc_id" style="width:100%" /></p> <textarea id="q" placeholder="输入你的问题"></textarea> <p><button id="send">提问</button></p> <div id="answer"></div> <script> const answerEl = document.getElementById("answer"); document.getElementById("send").onclick = async () => { const key = document.getElementById("key").value.trim(); const docId = document.getElementById("doc").value.trim(); const question = document.getElementById("q").value.trim(); if (!key || !docId || !question) return; answerEl.textContent = ""; const resp = await fetch("/api/chat", { method: "POST", headers: { "Content-Type": "application/json", "Authorization": "Bearer " + key }, body: JSON.stringify({ doc_id: docId, question }) }); if (!resp.ok) { answerEl.textContent = "请求失败：" + resp.status + " " + (await resp.text()); return; } const reader = resp.body.getReader(); const decoder = new TextDecoder("utf-8"); while (true) { const { done, value } = await reader.read(); if (done) break; answerEl.textContent += decoder.decode(value, { stream: true }); } }; </script> </body> </html>

前端这段代码的关键在decoder.decode(value, { stream: true })。中文是多字节编码，一个字符可能被切在两个分片之间，不加stream: true会出现乱码。这是接流式接口时非常高频的坑。

5.5 运行与验证

安装依赖并启动：

python -m venv .venv source .venv/bin/activate # Windows: .venv\Scripts\activate pip install -r requirements.txt uvicorn app.main:app --reload --port 8000

上传文档并取回doc_id：

curl -X POST http://127.0.0.1:8000/api/docs \ -H "Authorization: Bearer demo-key-001" \ -F "file=@./test.txt"

预期返回类似{"doc_id":"9f2c3a1b8d7e4f60","chunks":12}。然后用这个doc_id提问：

curl -N -X POST http://127.0.0.1:8000/api/chat \ -H "Authorization: Bearer demo-key-001" \ -H "Content-Type: application/json" \ -d '{"doc_id":"9f2c3a1b8d7e4f60","question":"这个产品的退款政策是什么？"}'

-N是关闭 curl 的缓冲，不加的话你会看到结果一次性跳出来，误以为流式没生效。

查询当日用量：

curl http://127.0.0.1:8000/api/usage -H "Authorization: Bearer demo-key-001"

把DAILY_QUOTA改成很小的值（比如 2），连续请求三次，第三次应该返回 429。这个验证动作建议每次都做，因为配额逻辑一旦出错，问题往往要到账单出来才发现。

6. 容器化部署与上线

能本地跑通和能上线是两件事。上线要做的最小集合：镜像打包、环境变量注入、反向代理、日志落地。

# 文件路径：Dockerfile FROM python:3.11-slim WORKDIR /app ENV PYTHONDONTWRITEBYTECODE=1 \ PYTHONUNBUFFERED=1 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY app ./app COPY web ./web RUN mkdir -p data/docs EXPOSE 8000 # 生产环境建议关闭 --reload，并按 CPU 核数调整 worker 数量 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "2"]

PYTHONUNBUFFERED=1不是可有可无的。不加这个，Python 的 stdout 会缓冲，容器日志会延迟很久才出现，排查线上问题时非常难受。

--workers设为 2 而不是默认的 1，是因为模型调用是 IO 密集型，单进程会浪费等待时间。但也不要盲目调大：如果你的文档存的是本地文件、配额用的是 SQLite，多 worker 会带来并发写问题。这类共享状态要么改用外部存储，要么就把 worker 数控制住并在部署上做单实例约束。

# 文件路径：docker-compose.yml services: app: build: . restart: unless-stopped env_file: - .env environment: DB_PATH: /app/data/app.db volumes: - ./data:/app/data ports: - "127.0.0.1:8000:8000" logging: driver: json-file options: max-size: "10m" max-file: "3"

几个细节：ports绑定到127.0.0.1而不是0.0.0.0，意味着这个端口不能从公网直接访问，必须经过前置的反向代理，这是最基本的安全边界；volumes把数据目录挂出来，容器重建时数据不丢；logging限制日志大小，避免磁盘被写满——这个坑非常经典，很多小服务跑几个月后磁盘告警，原因就是日志没有轮转。

上线前的检查清单，建议逐条打勾：

7. 常见问题与排查清单

排查这类问题有一个通用顺序：先看日志和状态码，区分是请求没进来、进来了报错、还是调通了但结果不对；然后在关键路径上加打印，确认是检索环节错了还是模型环节错了；最后用一条固定输入复现，改一个变量再验证。最忌讳的是同时改三四个地方，然后不知道是哪一处起了作用。

8. 工程最佳实践

8.1 密钥与权限

所有密钥走环境变量或密钥管理服务，不进代码、不进镜像、不进日志。日志里打印用户密钥时要脱敏，只留前几位。对外提供的 API Key 要做可撤销设计，也就是能单独作废某个用户的密钥而不影响其他人。权限遵循最小化原则：业务进程用的模型密钥只给调用权限，不要用主账号密钥。

8.2 先建评估集，再优化效果

这是我认为最被低估的一条。很多人调 Prompt 的方式是"改一改，感觉好像好了"。这种做法在两周后就会失控，因为你已经不记得哪次改动让哪类问题变坏了。

正确做法是先攒 30～50 条真实问题，每条标注期望答案的关键信息（不必逐字一致，标注要点即可）。之后每次改提示、改切块、换模型，都跑一遍这套用例，记录通过数。有了这个数字，你才知道改动是进步还是退步。这份评估集不需要什么平台，一个 JSON 文件加一个脚本就够。

8.3 成本与延迟优化

几个见效最快的方向：给max_tokens设一个合理上限，很多超额成本来自模型输出了一堆没人看的废话；召回条数不要贪多，top_k从 4 降到 3 往往对效果影响很小，但能省下不少上下文；对高频重复问题加一层结果缓存，命中率通常比想象的高；把非核心步骤（比如意图分类）用小模型，核心生成用大模型。

延迟方面，流式返回是最有效的体感优化——总耗时没变，但首字时间大幅缩短，用户感知完全不同。

8.4 可观测性

至少记录四类信息：请求标识、用户标识（脱敏）、耗时、token 消耗或调用次数。有了这四项，你才能回答"今天为什么慢""哪个用户消耗最多""错误集中在哪个环节"。

不要只记成功请求。失败的请求往往信息量更大——超时、限流、解析失败，这些才是需要优化的地方。

8.5 数据与合规边界

上传的文档可能包含敏感信息。工程上要做三件事：明确告知用户数据用途并取得授权；提供删除接口，且删除要真正生效（文件、索引、缓存都要清）；不要把用户数据用于与约定不符的用途。如果涉及个人信息或商业机密，需要在产品层面就做好隔离设计，而不是事后补救。

9. 学习路线与阶段目标

最后给一个可执行的时间表。下面这个节奏按业余时间每天投入 2 小时估算，全职学习可以压缩到一半左右。它不是唯一路径，但顺序经过验证是合理的。

几个执行建议。

第一，每个阶段必须有交付物，不要只看不写。看教程的时候觉得都懂，自己写的时候才发现每个环节都有细节。第 1 周的"三接口小服务"看起来很简单，但很多人卡在请求模型校验和错误码这两件事上。

第二，第 5 周之后再考虑换向量检索。前面几周用轻量打分方案，是为了让你清楚"检索质量差"到底差在哪一步。直接上重型方案，出问题时你会怀疑是切块错了、模型错了还是库用错了，根本无从下手。

第三，不要过早引入 Agent 框架。框架能省代码，但会掩盖调用链。等你手工把一条完整的调用链跑通、知道每一步的输入输出长什么样，再用框架加速，才是正确顺序。

第四，找真实文档练手。用一两段自己编的文本做测试，效果永远好看。找一份几十页的真实产品手册、公开的技术文档，跑一遍，你会立刻发现切块策略和召回质量的问题。这些问题是编造的测试数据里永远碰不到的。

回到开头那句话：薪资是结果，不是路径。这条路径上真正值钱的不是"会调用某个模型的接口"，而是"能在效果、成本、稳定性之间做权衡并给出可验证的方案"。上面这个文档问答项目虽然小，但它把鉴权、配额、检索、流式、部署、评估这几个关键环节全串了一遍，任何一个环节你都能继续深挖下去——比如把配额换成真正的订阅计费、把检索换成混合召回、把评估集扩到几百条做成自动化回归。把这些做扎实，你手上就不只是几个 demo，而是一套能直接复用到真实项目里的工程能力。

如果这篇文章对你有帮助，可以收藏备用；代码部分建议先原样跑通一遍，再按自己的业务往里面加功能，改动过程中遇到问题，回头看第 7 节的排查表，大部分场景都能对上。