第6章 实战项目：端到端 AI 应用
前五章我们分别学会了调 API、做 RAG、微调模型、搭 Agent - 但它们还是一个一个孤立的"零件"。这一章，我们把所有零件装进一台完整的机器：一个能回答公司文档问题、还能查库存查订单、最终部署上线的企业知识库问答机器人。你会跟着走完从需求拆解、架构设计、索引构建、问答服务、Agent 技能到部署上线、持续迭代的全过程，得到的是一个能放进简历、也能放进生产的真项目。
6.1 项目选型与需求拆解
学到这里，你手里已经有了调 API、RAG、微调、Agent 四张牌，但单独哪一张都还不是一个"产品"。这一章我们选一个真实、常见、技术覆盖最全的场景来练兵：企业知识库问答机器人 - 员工在对话框里提问，机器人检索公司文档、给出带引用的回答，再逐步升级成能查库存、查订单的"办事助理"。选它的理由有三：需求真实（几乎每家公司都需要内部知识问答）；技术覆盖全（检索、对话、工具调用、部署、评估全用上）；成果可演示、可上线（两周内能做出一个敢给同事用的版本）。
动手写代码之前，先把需求讲清楚。用一句话定义这个项目："员工提问 → 机器人查公司文档 → 返回带引用的回答"。拆开是三个动作，对应三个能力点：检索要对（能找到对的文档）、生成要稳（只依据资料回答、不编造）、引用要真（每条结论都能回到原文）。再把这句话展开成一张功能清单，每一行都配上可验收的标准：
| 功能模块 | 用户故事 | 验收标准 |
|---|---|---|
| 文档问答 | "差旅报销的流程是什么？"能基于制度文档回答 | 抽样 50 个问题，答对率 ≥ 90% |
| 引用溯源 | 回答需注明来源，能回到原文段落 | 100% 的回答带引用，点击可跳转 |
| 多轮对话 | 追问"那发票呢"能结合上文理解 | 连续追问 3 轮上下文不丢、不串题 |
| 拒答兜底 | 问知识库外的问题 | 明确说"不知道"，绝不编造 |
| Agent 技能（v2） | "查一下 A 商品的库存" | 结果与库存系统一致，5 秒内返回 |
这张表就是项目的"合同"。"先想清楚做什么，再写代码"：功能清单里的每一行，既是和需求方对齐的验收标准，也是上线前的测试用例。答对率怎么抽？引用能不能点？追问三句会不会丢上下文？把这些先定死，开发时每完成一项就照着验收一次，比写完再返工高效得多。
图 6-1 需求 → 功能 → 验收：先讲清楚"做什么"，再动手写代码
MVP 节奏：控制第一版范围
不要一上来就做全部。v1 只做"文档问答 + 引用溯源"，跑通、上线、积累数据；"Agent 技能"放到 v2（6.5 节）再上。范围越小，上线越快，迭代越快 - 第一版被同事用起来，比第一版功能全更重要。
6.2 系统架构设计
需求清楚了，接下来画架构。整个系统分成四层，每一层只干一件事，层与层之间通过明确的接口通信：前端聊天界面是用户看到的对话框；后端 API 服务（FastAPI）接收请求、做鉴权限流、记录日志，是所有流量的唯一入口；RAG 编排层是整个系统的"大脑"，负责意图判断、检索召回、组装 Prompt、调用模型生成、整理引用；最底层是两项资源 - 向量数据库（存文档向量）和大模型 API（负责生成）。
三个关键决策值得说清楚。第一，为什么把 RAG 单独抽成一层？因为它是全系统最常调整的部分：换切分参数、换向量库、加重排，都只动这一层，不动其他层。第二，为什么向量库独立部署？因为数据要和代码解耦 - 文档更新、索引重建都不需要重启服务，索引坏了也不影响服务本身。第三，为什么大模型走 API 而不是本地部署？省运维、按量付费，而且第 4 章微调出来的模型，随时可以替换到这里，对上层透明。
图 6-2 系统架构：前端 / API / RAG / 向量库 + 大模型 四层，Agent 工具层与离线索引管线虚线扩展
类比：架构像一家餐厅
前端像前台，收单传菜；API 服务像大堂经理，管排队和投诉；RAG 编排层像后厨 - 切配（检索）、下锅（生成）；向量库是食材库，大模型是主厨。菜单变了只改后厨，主厨换了客人也尝不出来 - 这就是"每层可替换"的价值。
三条架构原则
依赖单向（上层调下层，不反向）；每层可替换（换模型、换向量库都不动其他层）；日志贯穿全链路（每个请求带 request_id，从进来到返回全程可查，这是 6.7 节评估的数据来源）。
6.3 数据准备与索引构建
架构搭好了，但机器人"脑子里"还什么都没有。索引构建是 RAG 的地基，第 3 章我们讲过流程，这里把它落实成一条可重复执行的离线管线：收集 → 清洗 → 切分 → 向量化 → 入库。这条管线在文档更新时随时重跑，不需要动任何服务代码。
第一步收集：把散落在各处的资料统一归档进 docs/ 目录，PDF、Markdown、导出的网页都行，命名带上类型和主题（如 财务制度-差旅报销.md），方便后面追溯来源。第二步清洗：去掉页眉页脚、水印、无关表格，统一编码，把敏感信息（身份证号、手机号）脱敏后再入库 - 这一步既是质量要求，也是合规要求。
第三步切分是质量的关键，别无脑按固定字数切。先按文档结构走：章节标题是天然边界，段落之间按空行、句号逐级细分；每块 512 字左右、相邻块重叠 64 字，既保证语义完整，又不丢边界信息。切分时把"来源文件 + 章节标题"存进元数据 - 引用溯源就靠它。第四步向量化、第五步入库，用第 3 章的 Embedding 模型把每个分片变成向量写进向量库。注意给集合加版本号：每次重建索引生成新版本，检索效果变差随时回退，而不是默默覆盖旧的。
# build_index.py —— 离线索引管线：收集 → 切分 → 向量化 → 入库
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_chroma import Chroma
CHUNK_SIZE, CHUNK_OVERLAP = 512, 64 # 512 字一块、重叠 64 字：语义完整又不丢边界
# 1. 收集：PDF / Markdown / 网页统一收进 docs/ 目录
loader = DirectoryLoader("docs/", glob="**/*.md", loader_cls=TextLoader)
documents = loader.load()
print(f"共加载 {len(documents)} 个文档")
# 2. 切分：按结构逐级切，标题保留为元数据，引用溯源靠它
splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE, chunk_overlap=CHUNK_OVERLAP,
separators=["\n\n", "\n", "。", " "])
chunks = splitter.split_documents(documents)
# 3. 向量化 + 4. 入库：一条命令完成，集合带版本号便于回退
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
persist_directory="./db/faq_v2", # 版本号 v2：效果变差随时回退到 v1
)
print(f"已入库 {len(chunks)} 个分片")怎么验证切分质量
建完索引别急着写接口。先随机抽 20 个真实问题做检索测试，看返回的前 5 个分片是不是答非所问。这一步发现的问题（切分太碎、同义词搜不到），比上线之后才发现便宜一百倍 - 检索的账，永远在索引阶段算。
6.4 问答服务实现
索引就绪，现在写用户真正面对的东西 - 问答接口。核心流程四步：检索 → 组装 Prompt → 生成 → 返回引用。每一步都有讲究：检索不是把最相似的 5 块无脑丢给模型，要先做分数过滤 - 相似度低于阈值的分片宁可不给。模型没有资料可依据时，回答"不知道"比硬编一个答案好一万倍。
Prompt 组装是"防守"的关键：system 里写死三条规则 - 只依据资料回答、不编造、按 [1][2] 格式标注引用。把检索到的分片按编号拼进上下文，模型生成时自然会用编号指代来源，后端再把它翻译成结构化引用，前端就能渲染成可点击的链接。代码很短，整个接口不到 30 行：
# api.py —— FastAPI 问答服务：检索 → 组装 Prompt → 生成 → 返回引用
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI(title="企业知识库问答")
class ChatRequest(BaseModel):
question: str
history: list = [] # 多轮对话历史（第 2 章记忆管理）
PROMPT = """你是公司内部知识库助手。只依据下面的资料回答，不要编造。
资料：
{context}
问题：{question}
回答要求：先给结论，再分条说明，最后用 [1][2] 标注引用。"""
@app.post("/api/chat")
def chat(req: ChatRequest):
# 1. 检索 + 分数过滤：低于阈值的分片不配进 Prompt
chunks = vectorstore.similarity_search_with_score(req.question, k=5)
top = [c for c, s in chunks if s > 0.35]
if not top: # 兜底：答不了就诚实说"不知道"
return {"answer": "抱歉，知识库里没有找到相关资料。", "citations": []}
# 2. 组装 Prompt：分片按 [1][2] 编号拼进上下文
context = "\n\n".join(f"[{i+1}] {c.page_content}" for i, c in enumerate(top))
messages = [{"role": "system",
"content": PROMPT.format(context=context, question=req.question)}]
# 3. 生成
resp = client.chat.completions.create(model="deepseek-chat", messages=messages)
# 4. 返回答案 + 结构化引用（含原文片段与来源文件）
return {"answer": resp.choices[0].message.content,
"citations": [{"text": c.page_content[:80],
"source": c.metadata["source"]} for c in top]}接口返回的 JSON 里，answer 是给用户看的正文，citations 是给前端渲染引用用的结构化数据：
{
"answer": "差旅报销需先提交线上申请，审批通过后出差，结束后 30 天内提交票据。[1] 报销时限为出差结束后 30 天。[2]",
"citations": [
{"text": "差旅费报销流程：出差前提交申请，审批通过后出行……", "source": "docs/财务制度-差旅报销.md"},
{"text": "报销时限：出差结束后 30 天内提交票据，逾期需说明……", "source": "docs/报销细则.md"}
]
}前端拿到这个 JSON，把正文里的 [1][2] 渲染成超链接，点击跳到原文段落；把"抱歉，知识库里没有找到相关资料"渲染成一条诚实的兜底回复。多轮对话时把 history 一起传进来，用户追问"那发票呢"，机器人就知道上文在谈报销。要更接近 ChatGPT 的体验，把生成接口换成流式（SSE），首字延迟能压进 1 秒。
兜底比硬答更重要
企业场景里，一个自信满满的错误回答会让用户彻底失去信任；一句"知识库里没有，我帮你反馈给管理员"反而加分。阈值过滤就是这道防线 - 这条 if 分支，是全书最便宜的防幻觉代码，别删。
6.5 给机器人加 Agent 技能
到这里，机器人是个优秀的"文档顾问"：你说它听，它查资料回答。但用户很快会问出知识库答不了的问题 - "A 商品现在有货吗？""订单 20240315 到哪一步了？"这些答案不在文档里，在数据库里、在内部系统里。把机器人升级成"能办事的助理"，靠的正是第 5 章的工具调用。
思路完全一样：先给模型注册"工具说明书"（名字、描述、参数），模型在对话中判断该用哪个工具、生成参数，我们执行并观察结果，再让模型基于结果组织最终回答。区别只在于：这里的工具是真实的 - 查数据库、调内部 API，返回值直接决定答案。
# tools.py —— 给机器人注册"能办事"的工具（复用第 5 章工具调用思路）
TOOLS = [
{"type": "function", "function": {
"name": "query_inventory",
"description": "查询某商品当前库存数量",
"parameters": {"type": "object", "properties": {
"sku": {"type": "string", "description": "商品 SKU 编号"}},
"required": ["sku"]}}},
{"type": "function", "function": {
"name": "query_order",
"description": "按订单号查询订单状态",
"parameters": {"type": "object", "properties": {
"order_id": {"type": "string"}},
"required": ["order_id"]}}},
]
def query_inventory(sku: str) -> dict: # 工具实现：查内部数据库
row = db.execute("SELECT stock FROM inventory WHERE sku=?", (sku,)).fetchone()
return {"sku": sku, "stock": row[0] if row else 0}
def query_order(order_id: str) -> dict: # 工具实现：调内部 API
return internal_api.get(f"/orders/{order_id}").json()
def run_agent(question: str):
resp = client.chat.completions.create(
model="deepseek-chat",
messages=[{"role": "user", "content": question}],
tools=TOOLS, tool_choice="auto") # 模型自主决定要不要调工具
msg = resp.choices[0].message
if msg.tool_calls: # 观察：执行工具，把结果带回
fn = msg.tool_calls[0].function
result = globals()[fn.name](**json.loads(fn.arguments))
return f"工具返回：{result}" # 再让模型基于结果生成最终回答
return msg.content从"知识库问答"到"能办事的助理"，只差这一层工具。落地有两条纪律：一是权限收口 - Agent 只能调只读接口，所有写操作（改库存、发起审批）必须回到人工确认，绝不能让模型一句"帮我删掉这个订单"就真的删了；二是全程可审计 - 每次工具调用记日志：模型说了什么、调了哪个工具、返回了什么，一条不落。
知识库是"记忆"，工具是"手脚"
知识库回答"是什么"（制度、参数、流程），工具解决"怎么办"（库存多少、订单到哪、能不能办）。两者组合才是完整的 AI 助理 - 这也是第 5 章讲的 Agent 安全边界，在真实项目里的样子。
6.6 部署与成本
代码写完了，机器人要真正服务同事，还有最后一公里：部署。方案追求"一键"：Docker 打包 + docker-compose 编排。任何一台 Linux 服务器上，clone 仓库、填好环境变量、docker compose up -d，三分钟跑起来。以后的每次更新，重新 build 再 up，无痛升级。
密钥管理是部署的第一条红线：API Key 绝不写进代码、绝不打进镜像，运行时通过环境变量注入。镜像里只留一个空的占位变量，部署时在服务器 .env 文件里填真实值 - .env 不进版本库、加进 .gitignore。数据库路径、模型名等配置同样走环境变量，镜像本身不携带任何环境信息。
# Dockerfile —— 把服务打包成镜像（不含任何密钥）
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
COPY . .
ENV OPENAI_API_KEY="" # 运行时通过 -e 传入，绝不写死
EXPOSE 8000
CMD ["uvicorn", "api:app", "--host", "0.0.0.0", "--port", "8000"]
# docker-compose.yml —— 一键启动：服务 + 向量库
services:
kb-bot:
build: .
ports: ["8000:8000"]
environment:
OPENAI_API_KEY: ${OPENAI_API_KEY} # 从服务器 .env 注入
VECTOR_DB: ./db然后是成本账。这个项目的开销主要来自三块：大模型对话 API（大头）、Embedding 与向量库、一台跑服务的小服务器。按日 1000 次问答的体量粗算：
| 成本项 | 说明 | 月成本参考 |
|---|---|---|
| 大模型对话 API | 1000 问/天 × 平均 3k token/次（含检索上下文） | 约 300–900 元 |
| Embedding API | 增量索引为主，量很小 | 约 10–50 元 |
| 向量库 | 云托管版或自建（Chroma 可本地） | 0–200 元 |
| 云服务器 | 1 台 2C4G 跑 FastAPI + 数据库 | 约 100–300 元 |
| 合计 | 起步阶段（日 1000 问） | 约 500–1500 元/月 |
结论：起步阶段每月几百到一千多元，完全可接受。先小规模上线再扩容：第一周只开放给一个部门（比如财务部），用真实流量验证稳定性和回答质量，再逐步放开。用量上来之后，再上缓存、路由分层（简单问题走便宜小模型）、升级托管向量库，成本曲线就能一直压得住。
省钱三板斧
① 高频问题缓存：同一问句 7 天内直接回缓存，省掉整条 RAG 链路；② 分层路由：简单问题走小模型、复杂问题才上大模型；③ 控制检索条数：只把有用的 3–5 条分片放进 Prompt - Token 就是钱。
6.7 上线后的评估与迭代
上线不是终点，是迭代的起点。第一天就要把"数据"这件最值钱的事做起来：每个问题、机器人的回答、引用来源、用户有没有点"有帮助"，全部落库。没有这份日志，后面所有的"改进"都是拍脑袋 - 这也是架构图里"日志贯穿全链路"的真正用途。
每周固定一个时间做评估复盘。指标不用多，五六个就够：答对率、覆盖率、引用准确率、平均首字延迟、用户满意率。答对率怎么算？抽 50 个本周真实问题，人工打分，或让更强的模型当裁判逐条判定（第 7 章会讲 LLM-as-a-Judge）。
| 指标 | 定义 | 目标 |
|---|---|---|
| 答对率 | 抽样/评审判定回答正确的比例 | ≥ 90% |
| 覆盖率 | 知识库能回答的问题占全部提问的比例 | 逐步提升 |
| 引用准确率 | 引用来源与答案内容相符的比例 | ≥ 95% |
| 平均首字延迟 | 提问到返回首个字符的时间 | ≤ 3s |
| 用户满意率 | 好评 / (好评 + 差评) | ≥ 80% |
复盘之后是改进动作，最关键的机制是坏案例进测试集：本周答错的问题，一条条收进回归测试集，下次改完代码先跑一遍全量回归 - 改进 A 问题，绝不能弄坏 B 问题。然后针对 bad case 分类施策：检索不到，就调切分、补同义词、加重排；答错了，就修 Prompt、补文档；引用错，就查元数据。
图 6-3 持续改进的飞轮：记录 → 分析 → 沉淀 → 改进 → 评估 → 上线，循环往复
记录 → 分析 → 沉淀 → 改进 → 评估 → 上线，再回到记录 - 这个飞轮每转一圈，机器人就变好一点。三个月后回头看，它和你第一天上线时的样子会是两个产品。这就是 AI 应用和传统软件最大的不同：没有"做完"的一天，只有持续变好的每一天。
"AI 应用没有'做完'的一天，只有'更好'的一天。飞轮转起来，剩下的交给时间。"
本章要点
- 先需求后代码：功能清单 = 验收标准 = 测试用例，动手前先写清楚"做什么"。
- 四层架构：前端 / API / RAG / 向量库 + 大模型，依赖单向、每层可替换、日志贯穿全链路。
- 索引管线五步：收集 → 清洗 → 切分 → 向量化 → 入库，集合带版本号可回退。
- 问答四步：检索 → 组装 Prompt → 生成 → 返回引用；分数阈值过滤，答不了就说"不知道"。
- Agent 技能 = 工具调用：先只读后写，权限收口，全程可审计。
- 部署：Docker 一键起、密钥走环境变量、先小规模上线再扩容、成本按请求量粗算。
- 迭代飞轮：记录 → 分析 → 沉淀 → 改进 → 评估 → 上线，坏案例进测试集。