1. 为什么我要从零手搓一套 AI 工程化流程第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体框架而是一堆踩过的坑。过去两年我参与过三个从零起步的 AI 应用项目有做知识库问答的有做文档结构化抽取的也有做多轮对话客服的。每次项目启动团队里总有人提议“直接上某某平台”“用某某云的一站式方案”结果往往是前期跑得飞快到了中期开始被各种黑盒行为卡住调不动、改不了、成本还压不下来。所以当我决定认真梳理一套“从零构建 AI 工程”的方法论时我的出发点很朴素把每一个环节都拆开搞清楚它到底在干什么然后自己动手实现一遍最小可用版本。这套东西不是要替代成熟框架而是让你在真正用框架之前先建立对底层机制的直觉。就像学做菜你可以买现成的料理包但如果你连火候、油温、调味比例都没概念出了问题只能干瞪眼。ai-engineering-from-scratch这个主题核心就是回答一个问题一个 AI 应用从想法到上线中间到底要经过哪些工程环节每个环节的关键决策点在哪里。它适合那些不满足于“调个 API 就完事”的开发者也适合带团队的技术负责人——你需要知道哪些地方可以偷懒哪些地方偷懒会出大事。我打算按我自己实际搭建的顺序来讲先想清楚整体架构再逐个模块手写实现最后把整条链路串起来跑通。中间会穿插大量我踩过的坑和实测有效的参数。文章会比较长因为这件事本身就不短。2. 整体架构设计与技术选型思路2.1 从需求倒推架构先画数据流再选组件很多人一上来就纠结“用 LangChain 还是 LlamaIndex”我觉得这是本末倒置。正确的顺序是先把数据从输入到输出的完整路径画出来标出每一步的输入输出格式然后再看每个节点用什么工具实现最顺手。以我最近做的一个文档问答项目为例数据流大致是这样的用户上传 PDF 或 Word 文档文档解析成纯文本保留段落结构文本按语义切分成块每个块通过嵌入模型转成向量向量存入向量数据库同时保留原文映射用户提问问题也转成向量在向量库中检索最相似的若干块把检索结果和问题拼成提示词送给大模型大模型生成回答返回给用户这条链路里每一步都有多种实现方式。比如文档解析你可以用pdfplumber、PyMuPDF、unstructured也可以直接调云服务。选哪个取决于你的文档类型、预算和对解析质量的容忍度。我一般会先用PyMuPDF快速跑一遍看看解析出来的文本质量如果表格和公式多再考虑上unstructured或者自己写规则。注意不要一上来就追求“全自动完美解析”。实际项目中80% 的文档用简单工具就能处理得不错剩下 20% 的硬骨头再单独优化。先跑通链路再逐个击破。2.2 为什么我坚持自己写检索层而不是直接用框架封装LangChain 和 LlamaIndex 都提供了现成的检索接口一行代码就能搞定“向量化检索拼接”。但我建议至少在第一个项目里自己手写一遍检索层。原因有三个第一你需要知道相似度到底是怎么算的。余弦相似度、点积、欧氏距离在不同归一化条件下结果差异很大。框架帮你选了默认值但你不清楚它为什么这么选出了问题就无从下手。第二检索质量直接决定最终回答质量。我见过太多项目模型换了又换提示词调了又调最后发现是检索出来的内容根本不对。自己写检索层你可以在每一步加日志、加评估快速定位问题。第三成本控制。嵌入模型有按 token 收费的也有本地跑的。检索时的 top-k 设多少要不要做重排序这些决策直接影响账单。自己写一遍你心里有数。下面是我手写检索层的核心代码结构用 Python 写的依赖只有numpy和openai用于调嵌入模型import numpy as np from openai import OpenAI client OpenAI() def get_embedding(text, modeltext-embedding-3-small): text text.replace(\n, ) return client.embeddings.create(input[text], modelmodel).data[0].embedding def cosine_similarity(a, b): a np.array(a) b np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def retrieve(query, chunks, chunk_embeddings, top_k5): query_emb get_embedding(query) scores [cosine_similarity(query_emb, emb) for emb in chunk_embeddings] ranked np.argsort(scores)[::-1][:top_k] return [(chunks[i], scores[i]) for i in ranked]这段代码很朴素但它把“嵌入-存储-检索”的每一步都暴露出来了。你可以清楚地看到嵌入是逐条调用的相似度是逐个算的排序是简单的 argsort。在实际项目中我会把嵌入改成批量调用把相似度计算向量化但逻辑不变。2.3 嵌入模型选型别只看排行榜要看你的数据分布选嵌入模型时很多人直接看 MTEB 排行榜挑排名最高的。但排行榜上的评测数据集和你的实际数据往往分布不同。我的经验是用你自己的数据做一个小规模评测比看任何排行榜都靠谱。具体做法从你的文档里挑 20 个问题每个问题人工标注 3-5 个最相关的段落。然后用候选嵌入模型跑一遍检索看 top-5 里命中了几个人工标注的相关段落。这个指标叫 Recall5比排行榜上的综合分数更能反映实际效果。我实测下来对于中文技术文档text-embedding-3-small和bge-large-zh的表现差距不大但后者可以本地部署长期成本更低。如果你的文档以英文为主text-embedding-3-small的性价比很高。如果预算充足且对精度要求极高可以上text-embedding-3-large但要注意它的向量维度是 3072存储和计算成本都会上去。提示嵌入模型的维度直接影响向量数据库的存储和检索速度。1536 维和 3072 维在百万级数据量下检索延迟可能差一倍。选型时要考虑你的数据规模和延迟要求。2.4 向量数据库从内存版开始别急着上集群向量数据库的选择很多Pinecone、Weaviate、Qdrant、Milvus、Chroma、FAISS。我的建议是开发阶段直接用 FAISS 或 Chroma 的内存模式数据量超过 10 万条再考虑持久化方案。原因很简单内存版没有网络开销调试方便你可以随时打印出向量和索引状态。等到数据量上来了再把存储层换成 Qdrant 或 Milvus检索逻辑几乎不用改。我用 FAISS 的IndexFlatIP内积索引配合归一化向量来实现余弦相似度检索代码大概是这样import faiss import numpy as np dimension 1536 index faiss.IndexFlatIP(dimension) # 假设 embeddings 是归一化后的向量矩阵 embeddings np.array(chunk_embeddings).astype(float32) faiss.normalize_L2(embeddings) index.add(embeddings) # 检索 query_vec np.array([get_embedding(query)]).astype(float32) faiss.normalize_L2(query_vec) distances, indices index.search(query_vec, top_k)这里的关键是faiss.normalize_L2它把向量归一化到单位长度这样内积就等于余弦相似度。很多人忘了这一步导致检索结果和预期不符。3. 核心模块手写实现与关键细节3.1 文档解析别小看格式转换的坑文档解析是整条链路的第一环也是最容易被低估的一环。我遇到过 PDF 里的文字顺序错乱、Word 里的表格变成一堆乱码、扫描件根本没有文字层等问题。下面是我总结的常见格式处理策略格式推荐工具注意事项纯文本直接读取注意编码统一转 UTF-8PDF文字层PyMuPDF按页提取保留段落换行PDF扫描件OCR 工具需要额外处理准确率受图像质量影响Wordpython-docx表格和段落要分开处理Markdown直接读取注意代码块和表格的保留HTMLBeautifulSoup去掉导航和广告只留正文对于 PDF我一般用 PyMuPDF 的get_text(blocks)方法它返回的是文本块列表每个块带有坐标信息。这样我可以根据坐标判断哪些块属于同一段落避免把页眉页脚混进来。import fitz def parse_pdf(file_path): doc fitz.open(file_path) paragraphs [] for page in doc: blocks page.get_text(blocks) for block in blocks: x0, y0, x1, y1, text, block_no, block_type block if block_type 0: # 文本块 text text.strip() if text: paragraphs.append(text) return paragraphs这段代码会返回一个段落列表。实际使用时我还会加一些过滤规则比如去掉长度小于 10 个字符的块去掉包含“第 X 页”的块。注意PyMuPDF 的get_text(blocks)返回的文本块顺序不一定和阅读顺序一致。对于多栏排版的 PDF可能需要根据坐标重新排序。我一般会先按 y 坐标排序再按 x 坐标排序模拟从上到下、从左到右的阅读顺序。3.2 文本切分固定长度 vs 语义切分文本切分直接影响检索质量。切得太碎上下文丢失切得太长噪声太多。我试过三种策略固定长度切分按字符数或 token 数切简单粗暴。优点是实现快缺点是可能把一句话切成两半。我一般会设置 10%-20% 的重叠缓解边界问题。按段落切分以自然段落为单位超长段落再按句子切。优点是语义完整缺点是段落长度不均有的段落可能只有一句话。语义切分用嵌入模型计算相邻句子的相似度在相似度骤降的地方切分。优点是切分点更合理缺点是计算成本高。我现在的默认方案是先按段落切分如果段落超过 500 个 token再按句子切分句子之间保留 1-2 句的重叠。这个策略在大多数场景下够用实现也不复杂。import re def split_by_sentence(text, max_tokens500, overlap_sentences1): sentences re.split(r(?[。.!?])\s*, text) chunks [] current_chunk [] current_length 0 for sentence in sentences: sentence_length len(sentence) if current_length sentence_length max_tokens and current_chunk: chunks.append(.join(current_chunk)) # 保留重叠句子 current_chunk current_chunk[-overlap_sentences:] if overlap_sentences 0 else [] current_length sum(len(s) for s in current_chunk) current_chunk.append(sentence) current_length sentence_length if current_chunk: chunks.append(.join(current_chunk)) return chunks这里的max_tokens我一般设 500因为大多数嵌入模型的最大输入长度是 512 个 token。重叠句子数设 1-2保证边界处的语义连贯。3.3 提示词组装把检索结果变成模型能理解的上下文检索出相关段落后需要把它们和用户问题拼成提示词。这一步看似简单但有几个细节很关键第一给每个段落编号。这样模型在回答时可以引用来源比如“根据文档第 3 段”。我一般用[1]、[2]这样的格式。第二明确指令。告诉模型“只根据以下文档回答问题如果文档中没有相关信息就说不知道”。这能有效减少幻觉。第三控制总长度。检索结果加上问题总 token 数不能超过模型的上下文窗口。我一般会预留 20% 的空间给模型生成回答。下面是我常用的提示词模板你是一个基于文档回答问题的助手。请根据以下文档片段回答用户问题。 文档片段 [1] {chunk_1} [2] {chunk_2} ... 用户问题{query} 要求 1. 只根据上述文档片段回答不要编造信息。 2. 如果文档中没有相关信息直接说“根据现有文档无法回答”。 3. 回答时引用相关片段的编号如 [1]。这个模板我用了很多次实测下来能显著降低幻觉率。特别是第 2 条要求让模型在不知道的时候敢于说不知道比强行编一个答案要好得多。3.4 大模型调用参数调优与成本控制调用大模型时有几个参数需要重点关注temperature控制随机性。做问答时我一般设 0 或 0.1保证回答稳定。做创意生成时可以设 0.7-0.9。max_tokens限制生成长度。设太小可能回答不完整设太大浪费成本。我一般根据问题类型设 500-1000。top_p另一种控制随机性的方式。我一般和 temperature 二选一不同时调。频率惩罚和存在惩罚用于减少重复。问答场景一般不需要调。成本控制方面我做了两件事一是缓存嵌入结果同样的文本不重复调用嵌入接口二是对检索结果做去重避免把相似度高的重复段落都送给模型。import hashlib import json cache {} def get_embedding_cached(text, modeltext-embedding-3-small): key hashlib.md5(f{model}:{text}.encode()).hexdigest() if key in cache: return cache[key] embedding get_embedding(text, model) cache[key] embedding return embedding这个缓存很简单但在开发阶段能省不少钱。生产环境可以把缓存换成 Redis支持多进程共享。4. 完整实操流程从零跑通一个文档问答系统4.1 环境准备与依赖安装我假设你用的是 Python 3.10操作系统不限。先创建一个虚拟环境然后安装依赖python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install openai faiss-cpu numpy pymupdf python-docx如果你要用本地嵌入模型还需要安装sentence-transformerspip install sentence-transformers我建议先用 OpenAI 的嵌入接口跑通流程因为不需要下载模型速度快。等流程跑通了再考虑换成本地模型。4.2 第一步解析文档并切分假设你有一个docs文件夹里面放着 PDF 和 Word 文档。下面的代码会遍历所有文档解析成文本然后切分成块import os import fitz from docx import Document def parse_document(file_path): ext os.path.splitext(file_path)[1].lower() if ext .pdf: return parse_pdf(file_path) elif ext .docx: return parse_docx(file_path) elif ext .txt: with open(file_path, r, encodingutf-8) as f: return [f.read()] else: return [] def parse_docx(file_path): doc Document(file_path) paragraphs [p.text.strip() for p in doc.paragraphs if p.text.strip()] return paragraphs def load_all_documents(docs_dir): all_chunks [] for filename in os.listdir(docs_dir): file_path os.path.join(docs_dir, filename) if os.path.isfile(file_path): paragraphs parse_document(file_path) for para in paragraphs: chunks split_by_sentence(para) all_chunks.extend(chunks) return all_chunks这段代码会把所有文档的段落收集起来然后按句子切分。实际使用时你可能需要根据文档类型调整切分策略。4.3 第二步生成嵌入并建立索引拿到所有文本块后批量生成嵌入向量然后建立 FAISS 索引def build_index(chunks): embeddings [] for chunk in chunks: emb get_embedding_cached(chunk) embeddings.append(emb) embeddings np.array(embeddings).astype(float32) faiss.normalize_L2(embeddings) dimension embeddings.shape[1] index faiss.IndexFlatIP(dimension) index.add(embeddings) return index, embeddings这里我用了缓存函数避免重复调用嵌入接口。faiss.normalize_L2把向量归一化这样内积检索就等于余弦相似度。4.4 第三步检索与回答生成用户提问时先检索相关段落再拼提示词调用大模型def answer_question(query, chunks, index, top_k5): query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800 ) return response.choices[0].message.content def build_prompt(query, chunks): chunk_text \n.join([f[{i1}] {chunk} for i, chunk in enumerate(chunks)]) return f你是一个基于文档回答问题的助手。请根据以下文档片段回答用户问题。 文档片段 {chunk_text} 用户问题{query} 要求 1. 只根据上述文档片段回答不要编造信息。 2. 如果文档中没有相关信息直接说“根据现有文档无法回答”。 3. 回答时引用相关片段的编号如 [1]。 这套流程跑下来一个最小可用的文档问答系统就成型了。你可以用streamlit或gradio快速搭一个界面方便测试。4.5 第四步效果评估与迭代跑通之后别急着上线。先做一轮评估看看检索和回答的质量。我的做法是准备 20-30 个测试问题覆盖文档的主要主题。对每个问题人工标注 3-5 个相关段落。跑检索计算 Recall5 和 MRR平均倒数排名。对回答质量打分1-5 分看平均分和低分案例。如果 Recall5 低于 0.8说明检索有问题需要调整切分策略或换嵌入模型。如果 Recall5 高但回答质量低说明提示词或模型有问题需要调整提示词模板或换模型。提示评估集要覆盖不同类型的问题包括事实型“X 是什么”、比较型“X 和 Y 的区别”、推理型“根据文档如果 X 发生Y 会怎样”。不同类型的问题对检索和生成的要求不同。5. 常见问题与排查技巧实录5.1 检索结果不相关从嵌入质量查起这是最常见的问题。用户问了一个问题检索出来的段落完全不沾边。排查顺序如下第一步检查嵌入是否正常。随便拿两个语义相似的句子算一下余弦相似度应该在 0.8 以上。如果低于 0.5说明嵌入模型有问题或者文本预处理有问题比如混入了大量特殊字符。第二步检查切分是否合理。如果切分太碎每个块只有几个词嵌入向量包含的信息太少检索自然不准。我一般会打印几个块的文本看看是否语义完整。第三步检查 top_k 是否太小。top_k3 可能不够试试调到 10看看相关段落是否出现在结果里。如果出现了说明排序有问题可以考虑加重排序模型。第四步检查查询和文档的语言是否一致。如果文档是中文查询是英文嵌入模型可能无法正确匹配。这种情况下需要做查询翻译或者用多语言嵌入模型。5.2 模型回答“不知道”检索没召回还是提示词太严模型说“根据现有文档无法回答”有两种可能一是检索确实没找到相关段落二是提示词太严格模型不敢回答。先看检索结果。如果检索出来的段落确实不相关那就是检索问题按上面的步骤排查。如果检索出来的段落相关但模型还是说不知道那就是提示词问题。可以试着把“如果文档中没有相关信息直接说不知道”改成“尽量根据文档回答如果信息不完整可以补充你的理解”看看模型是否愿意回答。我一般会保留严格的提示词因为宁可让模型说不知道也不要它编造答案。但如果发现模型过于保守可以适当放宽。5.3 回答太长或太短调整 max_tokens 和提示词回答长度不受控通常是max_tokens设得太大或太小。我一般设 800对于大多数问答够用。如果回答经常被截断调到 1200。如果回答太啰嗦可以在提示词里加一句“请简洁回答不超过 3 句话”。还有一个技巧在提示词里指定回答格式。比如“请用 bullet point 列出 3 个要点”这样模型会按格式输出长度自然受控。5.4 成本超预期从嵌入缓存和检索去重入手成本主要来自两块嵌入接口调用和模型生成。嵌入缓存能省掉重复文本的调用检索去重能减少送给模型的 token 数。去重的做法很简单检索出 top_k 个段落后计算两两之间的相似度如果超过 0.95只保留一个。这样能避免把几乎相同的段落都送给模型。def deduplicate_chunks(chunks, embeddings, threshold0.95): keep [] for i, emb in enumerate(embeddings): is_duplicate False for j in keep: if cosine_similarity(emb, embeddings[j]) threshold: is_duplicate True break if not is_duplicate: keep.append(i) return [chunks[i] for i in keep]这个函数在检索后调用能有效减少冗余。5.5 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关嵌入质量差算相似句子的余弦相似度换嵌入模型或清洗文本检索结果不相关切分太碎打印块文本调整切分策略增大块长度模型说不知道检索没召回检查 top_k 结果增大 top_k 或换嵌入模型模型说不知道提示词太严放宽提示词测试调整提示词措辞回答太长max_tokens 太大检查参数调小 max_tokens 或加格式要求回答太短max_tokens 太小检查参数调大 max_tokens成本超预期重复调用嵌入加日志统计加缓存和去重响应太慢检索或生成慢分步计时优化索引或换更快的模型6. 我踩过的坑和实测有效的经验6.1 别在切分上偷懒这是回报率最高的优化点我做过一个对比实验同一套文档同一套嵌入模型同一套提示词只改变切分策略。固定长度切分的 Recall5 是 0.72按段落切分是 0.81语义切分是 0.85。从 0.72 到 0.85提升非常明显而成本只是多写了几十行切分代码。所以我的建议是在切分上多花点时间比换更贵的模型划算得多。具体做法是先按段落切超长段落再按句子切句子之间保留重叠。如果效果还不够再考虑语义切分。6.2 嵌入模型不是越贵越好要看你的数据分布我试过用text-embedding-3-large替换text-embedding-3-small成本翻了 6 倍多但 Recall5 只提升了 0.02。对于我的数据中文技术文档这个提升完全不值。后来换成bge-large-zh本地部署Recall5 和text-embedding-3-small持平但成本几乎为零。所以选嵌入模型时一定要用自己的数据做评测。排行榜只能参考不能直接照搬。6.3 提示词里的“不知道”指令能省掉很多麻烦早期我做问答系统时没有加“如果不知道就说不知道”的指令结果模型经常编造答案。用户问一个文档里没有的问题模型也能编出一段看似合理的回答。后来加了这条指令幻觉率大幅下降。但要注意这条指令不能太生硬。我试过“如果文档中没有相关信息直接说不知道”模型有时候会过度保守明明有相关信息也说不知道。后来改成“只根据上述文档片段回答如果文档中没有相关信息直接说‘根据现有文档无法回答’”效果好很多。6.4 评估集要提前准备不要等上线了才想起来我见过太多项目开发阶段跑得飞快上线后用户反馈一堆问题但因为没有评估集根本不知道问题出在哪。我的做法是项目启动第一周就准备 20-30 个测试问题随着开发进展不断补充。评估集不需要很大但要有代表性。我一般会覆盖文档的主要章节每个章节出 2-3 个问题。问题类型要多样包括事实型、比较型、推理型。有了评估集每次改动都能快速验证效果避免盲目调参。6.5 日志要打全不然排查问题全靠猜AI 应用的链路很长从文档解析到最终回答中间经过多个模块。如果日志打不全出了问题根本不知道是哪一步的锅。我的做法是每个模块的输入输出都打日志关键参数也打。比如检索模块我会记录查询文本、查询嵌入向量前 10 维、top_k 结果、每个结果的相似度分数。生成模块我会记录提示词全文、模型参数、生成结果、token 消耗。这些日志在排查问题时非常有用。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def answer_question_with_logging(query, chunks, index, top_k5): logging.info(fQuery: {query}) query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) logging.info(fTop-{top_k} indices: {indices[0]}) logging.info(fTop-{top_k} distances: {distances[0]}) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) logging.info(fPrompt length: {len(prompt)}) response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800 ) answer response.choices[0].message.content logging.info(fAnswer: {answer}) logging.info(fToken usage: {response.usage}) return answer这套日志在开发阶段会输出很多信息但排查问题时能省下大量时间。生产环境可以调高日志级别只记录关键信息。6.6 本地嵌入模型部署省钱的代价是运维如果你决定用本地嵌入模型比如bge-large-zh需要额外考虑几件事模型下载、GPU 显存、推理速度、并发处理。我一般用sentence-transformers加载模型然后封装成一个批量推理函数from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-large-zh-v1.5) def get_embeddings_local(texts, batch_size32): embeddings model.encode(texts, batch_sizebatch_size, normalize_embeddingsTrue) return embeddings.tolist()normalize_embeddingsTrue会自动归一化省去了手动归一化的步骤。批量大小根据显存调整一般 32 或 64 比较合适。本地模型的优点是成本低、数据不出本地缺点是首次加载慢、需要 GPU、并发能力有限。如果你的数据量不大或者对成本敏感本地模型是不错的选择。6.7 重排序检索之后的第二道防线如果检索出来的 top_k 结果里相关段落排名靠后可以考虑加重排序模型。重排序的思路是先用嵌入检索出 top-20然后用一个更精细的模型对这 20 个结果重新打分取 top-5 送给大模型。常用的重排序模型有bge-reranker系列可以本地部署。重排序的代价是增加一次模型推理但能显著提升检索精度。我实测下来在 top-20 里加一层重排序Recall5 能提升 0.05-0.1。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-large) def rerank(query, chunks, top_k5): pairs [[query, chunk] for chunk in chunks] scores reranker.predict(pairs) ranked_indices np.argsort(scores)[::-1][:top_k] return [chunks[i] for i in ranked_indices]重排序模型比嵌入模型大推理速度慢一些但只对少量候选做推理总体延迟增加不多。6.8 流式输出提升用户体验的简单手段大模型生成回答需要时间如果等全部生成完再返回用户会等很久。流式输出可以让用户看到回答逐字出现体验好很多。OpenAI 的接口支持流式输出def answer_question_stream(query, chunks, index, top_k5): query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800, streamTrue ) for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content用yield逐块返回前端可以实时渲染。这个改动很小但用户体验提升明显。6.9 多轮对话维护上下文的关键是会话状态单轮问答跑通后下一步是多轮对话。多轮对话的关键是维护会话状态用户问了什么、模型答了什么、检索了哪些文档。我一般用一个列表保存对话历史每次生成时把历史也拼进提示词。但要注意对话历史不能无限增长否则 token 会爆。我一般只保留最近 3-5 轮对话更早的对话做摘要或者直接丢弃。class Conversation: def __init__(self, max_history5): self.history [] self.max_history max_history def add_turn(self, role, content): self.history.append({role: role, content: content}) if len(self.history) self.max_history * 2: self.history self.history[-self.max_history * 2:] def get_context(self): return self.history多轮对话的检索策略也需要调整有时候需要结合历史问题来检索而不是只看当前问题。我一般会把最近一轮的用户问题和当前问题拼起来做检索效果比只看当前问题好。6.10 部署上线从脚本到服务的最后一公里开发阶段用脚本跑没问题上线需要封装成服务。我一般用 FastAPI 搭一个简单的 HTTP 接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): query: str top_k: int 5 app.post(/ask) def ask(request: QueryRequest): answer answer_question(request.query, chunks, index, request.top_k) return {answer: answer}然后用uvicorn启动uvicorn main:app --host 0.0.0.0 --port 8000这个服务很简单但足够支撑小规模使用。如果需要更高并发可以考虑加缓存、加队列、用异步接口。但那是另一个话题了。6.11 安全与合规别让模型说出不该说的话AI 应用上线前一定要考虑安全与合规。我一般会做几件事一是加输入过滤拦截明显恶意的查询二是加输出过滤检查模型回答是否包含敏感内容三是加日志审计记录所有查询和回答方便追溯。输入过滤可以用关键词匹配也可以用一个小模型做分类。输出过滤类似。这些措施不能保证 100% 安全但能挡住大部分明显问题。注意安全与合规是底线不要为了追求效果而忽略。上线前一定要过一遍安全审查确保没有明显漏洞。6.12 持续迭代上线只是开始AI 应用上线后需要持续迭代。我一般会关注几个指标用户查询量、检索命中率、回答满意度、平均响应时间、token 消耗。根据这些指标定期优化切分策略、嵌入模型、提示词模板。用户反馈是最宝贵的优化线索。我一般会在界面上加一个“这个回答有帮助吗”的按钮收集用户反馈。负面反馈的查询我会重点分析看看是检索问题还是生成问题。这套流程跑下来一个从零构建的 AI 工程化文档问答系统就完整了。它不依赖任何重型框架每个环节都清晰可控。你可以根据实际需求替换其中的任何模块比如把 FAISS 换成 Qdrant把 OpenAI 嵌入换成 bge把 GPT-4o-mini 换成其他模型。核心思路不变理解每个环节在干什么然后自己动手实现一遍。我在实际项目中发现这套从零构建的经验让我在使用成熟框架时更加得心应手。因为我知道框架在背后做了什么哪些地方可以信任哪些地方需要自己接管。这种掌控感是直接调 API 永远给不了的。 从零手搓AI工程化流程：文档问答系统实战与避坑指南
1. 为什么我要从零手搓一套 AI 工程化流程第一次看到ai-engineering-from-scratch这个标题我脑子里蹦出来的不是某个具体框架而是一堆踩过的坑。过去两年我参与过三个从零起步的 AI 应用项目有做知识库问答的有做文档结构化抽取的也有做多轮对话客服的。每次项目启动团队里总有人提议“直接上某某平台”“用某某云的一站式方案”结果往往是前期跑得飞快到了中期开始被各种黑盒行为卡住调不动、改不了、成本还压不下来。所以当我决定认真梳理一套“从零构建 AI 工程”的方法论时我的出发点很朴素把每一个环节都拆开搞清楚它到底在干什么然后自己动手实现一遍最小可用版本。这套东西不是要替代成熟框架而是让你在真正用框架之前先建立对底层机制的直觉。就像学做菜你可以买现成的料理包但如果你连火候、油温、调味比例都没概念出了问题只能干瞪眼。ai-engineering-from-scratch这个主题核心就是回答一个问题一个 AI 应用从想法到上线中间到底要经过哪些工程环节每个环节的关键决策点在哪里。它适合那些不满足于“调个 API 就完事”的开发者也适合带团队的技术负责人——你需要知道哪些地方可以偷懒哪些地方偷懒会出大事。我打算按我自己实际搭建的顺序来讲先想清楚整体架构再逐个模块手写实现最后把整条链路串起来跑通。中间会穿插大量我踩过的坑和实测有效的参数。文章会比较长因为这件事本身就不短。2. 整体架构设计与技术选型思路2.1 从需求倒推架构先画数据流再选组件很多人一上来就纠结“用 LangChain 还是 LlamaIndex”我觉得这是本末倒置。正确的顺序是先把数据从输入到输出的完整路径画出来标出每一步的输入输出格式然后再看每个节点用什么工具实现最顺手。以我最近做的一个文档问答项目为例数据流大致是这样的用户上传 PDF 或 Word 文档文档解析成纯文本保留段落结构文本按语义切分成块每个块通过嵌入模型转成向量向量存入向量数据库同时保留原文映射用户提问问题也转成向量在向量库中检索最相似的若干块把检索结果和问题拼成提示词送给大模型大模型生成回答返回给用户这条链路里每一步都有多种实现方式。比如文档解析你可以用pdfplumber、PyMuPDF、unstructured也可以直接调云服务。选哪个取决于你的文档类型、预算和对解析质量的容忍度。我一般会先用PyMuPDF快速跑一遍看看解析出来的文本质量如果表格和公式多再考虑上unstructured或者自己写规则。注意不要一上来就追求“全自动完美解析”。实际项目中80% 的文档用简单工具就能处理得不错剩下 20% 的硬骨头再单独优化。先跑通链路再逐个击破。2.2 为什么我坚持自己写检索层而不是直接用框架封装LangChain 和 LlamaIndex 都提供了现成的检索接口一行代码就能搞定“向量化检索拼接”。但我建议至少在第一个项目里自己手写一遍检索层。原因有三个第一你需要知道相似度到底是怎么算的。余弦相似度、点积、欧氏距离在不同归一化条件下结果差异很大。框架帮你选了默认值但你不清楚它为什么这么选出了问题就无从下手。第二检索质量直接决定最终回答质量。我见过太多项目模型换了又换提示词调了又调最后发现是检索出来的内容根本不对。自己写检索层你可以在每一步加日志、加评估快速定位问题。第三成本控制。嵌入模型有按 token 收费的也有本地跑的。检索时的 top-k 设多少要不要做重排序这些决策直接影响账单。自己写一遍你心里有数。下面是我手写检索层的核心代码结构用 Python 写的依赖只有numpy和openai用于调嵌入模型import numpy as np from openai import OpenAI client OpenAI() def get_embedding(text, modeltext-embedding-3-small): text text.replace(\n, ) return client.embeddings.create(input[text], modelmodel).data[0].embedding def cosine_similarity(a, b): a np.array(a) b np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def retrieve(query, chunks, chunk_embeddings, top_k5): query_emb get_embedding(query) scores [cosine_similarity(query_emb, emb) for emb in chunk_embeddings] ranked np.argsort(scores)[::-1][:top_k] return [(chunks[i], scores[i]) for i in ranked]这段代码很朴素但它把“嵌入-存储-检索”的每一步都暴露出来了。你可以清楚地看到嵌入是逐条调用的相似度是逐个算的排序是简单的 argsort。在实际项目中我会把嵌入改成批量调用把相似度计算向量化但逻辑不变。2.3 嵌入模型选型别只看排行榜要看你的数据分布选嵌入模型时很多人直接看 MTEB 排行榜挑排名最高的。但排行榜上的评测数据集和你的实际数据往往分布不同。我的经验是用你自己的数据做一个小规模评测比看任何排行榜都靠谱。具体做法从你的文档里挑 20 个问题每个问题人工标注 3-5 个最相关的段落。然后用候选嵌入模型跑一遍检索看 top-5 里命中了几个人工标注的相关段落。这个指标叫 Recall5比排行榜上的综合分数更能反映实际效果。我实测下来对于中文技术文档text-embedding-3-small和bge-large-zh的表现差距不大但后者可以本地部署长期成本更低。如果你的文档以英文为主text-embedding-3-small的性价比很高。如果预算充足且对精度要求极高可以上text-embedding-3-large但要注意它的向量维度是 3072存储和计算成本都会上去。提示嵌入模型的维度直接影响向量数据库的存储和检索速度。1536 维和 3072 维在百万级数据量下检索延迟可能差一倍。选型时要考虑你的数据规模和延迟要求。2.4 向量数据库从内存版开始别急着上集群向量数据库的选择很多Pinecone、Weaviate、Qdrant、Milvus、Chroma、FAISS。我的建议是开发阶段直接用 FAISS 或 Chroma 的内存模式数据量超过 10 万条再考虑持久化方案。原因很简单内存版没有网络开销调试方便你可以随时打印出向量和索引状态。等到数据量上来了再把存储层换成 Qdrant 或 Milvus检索逻辑几乎不用改。我用 FAISS 的IndexFlatIP内积索引配合归一化向量来实现余弦相似度检索代码大概是这样import faiss import numpy as np dimension 1536 index faiss.IndexFlatIP(dimension) # 假设 embeddings 是归一化后的向量矩阵 embeddings np.array(chunk_embeddings).astype(float32) faiss.normalize_L2(embeddings) index.add(embeddings) # 检索 query_vec np.array([get_embedding(query)]).astype(float32) faiss.normalize_L2(query_vec) distances, indices index.search(query_vec, top_k)这里的关键是faiss.normalize_L2它把向量归一化到单位长度这样内积就等于余弦相似度。很多人忘了这一步导致检索结果和预期不符。3. 核心模块手写实现与关键细节3.1 文档解析别小看格式转换的坑文档解析是整条链路的第一环也是最容易被低估的一环。我遇到过 PDF 里的文字顺序错乱、Word 里的表格变成一堆乱码、扫描件根本没有文字层等问题。下面是我总结的常见格式处理策略格式推荐工具注意事项纯文本直接读取注意编码统一转 UTF-8PDF文字层PyMuPDF按页提取保留段落换行PDF扫描件OCR 工具需要额外处理准确率受图像质量影响Wordpython-docx表格和段落要分开处理Markdown直接读取注意代码块和表格的保留HTMLBeautifulSoup去掉导航和广告只留正文对于 PDF我一般用 PyMuPDF 的get_text(blocks)方法它返回的是文本块列表每个块带有坐标信息。这样我可以根据坐标判断哪些块属于同一段落避免把页眉页脚混进来。import fitz def parse_pdf(file_path): doc fitz.open(file_path) paragraphs [] for page in doc: blocks page.get_text(blocks) for block in blocks: x0, y0, x1, y1, text, block_no, block_type block if block_type 0: # 文本块 text text.strip() if text: paragraphs.append(text) return paragraphs这段代码会返回一个段落列表。实际使用时我还会加一些过滤规则比如去掉长度小于 10 个字符的块去掉包含“第 X 页”的块。注意PyMuPDF 的get_text(blocks)返回的文本块顺序不一定和阅读顺序一致。对于多栏排版的 PDF可能需要根据坐标重新排序。我一般会先按 y 坐标排序再按 x 坐标排序模拟从上到下、从左到右的阅读顺序。3.2 文本切分固定长度 vs 语义切分文本切分直接影响检索质量。切得太碎上下文丢失切得太长噪声太多。我试过三种策略固定长度切分按字符数或 token 数切简单粗暴。优点是实现快缺点是可能把一句话切成两半。我一般会设置 10%-20% 的重叠缓解边界问题。按段落切分以自然段落为单位超长段落再按句子切。优点是语义完整缺点是段落长度不均有的段落可能只有一句话。语义切分用嵌入模型计算相邻句子的相似度在相似度骤降的地方切分。优点是切分点更合理缺点是计算成本高。我现在的默认方案是先按段落切分如果段落超过 500 个 token再按句子切分句子之间保留 1-2 句的重叠。这个策略在大多数场景下够用实现也不复杂。import re def split_by_sentence(text, max_tokens500, overlap_sentences1): sentences re.split(r(?[。.!?])\s*, text) chunks [] current_chunk [] current_length 0 for sentence in sentences: sentence_length len(sentence) if current_length sentence_length max_tokens and current_chunk: chunks.append(.join(current_chunk)) # 保留重叠句子 current_chunk current_chunk[-overlap_sentences:] if overlap_sentences 0 else [] current_length sum(len(s) for s in current_chunk) current_chunk.append(sentence) current_length sentence_length if current_chunk: chunks.append(.join(current_chunk)) return chunks这里的max_tokens我一般设 500因为大多数嵌入模型的最大输入长度是 512 个 token。重叠句子数设 1-2保证边界处的语义连贯。3.3 提示词组装把检索结果变成模型能理解的上下文检索出相关段落后需要把它们和用户问题拼成提示词。这一步看似简单但有几个细节很关键第一给每个段落编号。这样模型在回答时可以引用来源比如“根据文档第 3 段”。我一般用[1]、[2]这样的格式。第二明确指令。告诉模型“只根据以下文档回答问题如果文档中没有相关信息就说不知道”。这能有效减少幻觉。第三控制总长度。检索结果加上问题总 token 数不能超过模型的上下文窗口。我一般会预留 20% 的空间给模型生成回答。下面是我常用的提示词模板你是一个基于文档回答问题的助手。请根据以下文档片段回答用户问题。 文档片段 [1] {chunk_1} [2] {chunk_2} ... 用户问题{query} 要求 1. 只根据上述文档片段回答不要编造信息。 2. 如果文档中没有相关信息直接说“根据现有文档无法回答”。 3. 回答时引用相关片段的编号如 [1]。这个模板我用了很多次实测下来能显著降低幻觉率。特别是第 2 条要求让模型在不知道的时候敢于说不知道比强行编一个答案要好得多。3.4 大模型调用参数调优与成本控制调用大模型时有几个参数需要重点关注temperature控制随机性。做问答时我一般设 0 或 0.1保证回答稳定。做创意生成时可以设 0.7-0.9。max_tokens限制生成长度。设太小可能回答不完整设太大浪费成本。我一般根据问题类型设 500-1000。top_p另一种控制随机性的方式。我一般和 temperature 二选一不同时调。频率惩罚和存在惩罚用于减少重复。问答场景一般不需要调。成本控制方面我做了两件事一是缓存嵌入结果同样的文本不重复调用嵌入接口二是对检索结果做去重避免把相似度高的重复段落都送给模型。import hashlib import json cache {} def get_embedding_cached(text, modeltext-embedding-3-small): key hashlib.md5(f{model}:{text}.encode()).hexdigest() if key in cache: return cache[key] embedding get_embedding(text, model) cache[key] embedding return embedding这个缓存很简单但在开发阶段能省不少钱。生产环境可以把缓存换成 Redis支持多进程共享。4. 完整实操流程从零跑通一个文档问答系统4.1 环境准备与依赖安装我假设你用的是 Python 3.10操作系统不限。先创建一个虚拟环境然后安装依赖python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate pip install openai faiss-cpu numpy pymupdf python-docx如果你要用本地嵌入模型还需要安装sentence-transformerspip install sentence-transformers我建议先用 OpenAI 的嵌入接口跑通流程因为不需要下载模型速度快。等流程跑通了再考虑换成本地模型。4.2 第一步解析文档并切分假设你有一个docs文件夹里面放着 PDF 和 Word 文档。下面的代码会遍历所有文档解析成文本然后切分成块import os import fitz from docx import Document def parse_document(file_path): ext os.path.splitext(file_path)[1].lower() if ext .pdf: return parse_pdf(file_path) elif ext .docx: return parse_docx(file_path) elif ext .txt: with open(file_path, r, encodingutf-8) as f: return [f.read()] else: return [] def parse_docx(file_path): doc Document(file_path) paragraphs [p.text.strip() for p in doc.paragraphs if p.text.strip()] return paragraphs def load_all_documents(docs_dir): all_chunks [] for filename in os.listdir(docs_dir): file_path os.path.join(docs_dir, filename) if os.path.isfile(file_path): paragraphs parse_document(file_path) for para in paragraphs: chunks split_by_sentence(para) all_chunks.extend(chunks) return all_chunks这段代码会把所有文档的段落收集起来然后按句子切分。实际使用时你可能需要根据文档类型调整切分策略。4.3 第二步生成嵌入并建立索引拿到所有文本块后批量生成嵌入向量然后建立 FAISS 索引def build_index(chunks): embeddings [] for chunk in chunks: emb get_embedding_cached(chunk) embeddings.append(emb) embeddings np.array(embeddings).astype(float32) faiss.normalize_L2(embeddings) dimension embeddings.shape[1] index faiss.IndexFlatIP(dimension) index.add(embeddings) return index, embeddings这里我用了缓存函数避免重复调用嵌入接口。faiss.normalize_L2把向量归一化这样内积检索就等于余弦相似度。4.4 第三步检索与回答生成用户提问时先检索相关段落再拼提示词调用大模型def answer_question(query, chunks, index, top_k5): query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800 ) return response.choices[0].message.content def build_prompt(query, chunks): chunk_text \n.join([f[{i1}] {chunk} for i, chunk in enumerate(chunks)]) return f你是一个基于文档回答问题的助手。请根据以下文档片段回答用户问题。 文档片段 {chunk_text} 用户问题{query} 要求 1. 只根据上述文档片段回答不要编造信息。 2. 如果文档中没有相关信息直接说“根据现有文档无法回答”。 3. 回答时引用相关片段的编号如 [1]。 这套流程跑下来一个最小可用的文档问答系统就成型了。你可以用streamlit或gradio快速搭一个界面方便测试。4.5 第四步效果评估与迭代跑通之后别急着上线。先做一轮评估看看检索和回答的质量。我的做法是准备 20-30 个测试问题覆盖文档的主要主题。对每个问题人工标注 3-5 个相关段落。跑检索计算 Recall5 和 MRR平均倒数排名。对回答质量打分1-5 分看平均分和低分案例。如果 Recall5 低于 0.8说明检索有问题需要调整切分策略或换嵌入模型。如果 Recall5 高但回答质量低说明提示词或模型有问题需要调整提示词模板或换模型。提示评估集要覆盖不同类型的问题包括事实型“X 是什么”、比较型“X 和 Y 的区别”、推理型“根据文档如果 X 发生Y 会怎样”。不同类型的问题对检索和生成的要求不同。5. 常见问题与排查技巧实录5.1 检索结果不相关从嵌入质量查起这是最常见的问题。用户问了一个问题检索出来的段落完全不沾边。排查顺序如下第一步检查嵌入是否正常。随便拿两个语义相似的句子算一下余弦相似度应该在 0.8 以上。如果低于 0.5说明嵌入模型有问题或者文本预处理有问题比如混入了大量特殊字符。第二步检查切分是否合理。如果切分太碎每个块只有几个词嵌入向量包含的信息太少检索自然不准。我一般会打印几个块的文本看看是否语义完整。第三步检查 top_k 是否太小。top_k3 可能不够试试调到 10看看相关段落是否出现在结果里。如果出现了说明排序有问题可以考虑加重排序模型。第四步检查查询和文档的语言是否一致。如果文档是中文查询是英文嵌入模型可能无法正确匹配。这种情况下需要做查询翻译或者用多语言嵌入模型。5.2 模型回答“不知道”检索没召回还是提示词太严模型说“根据现有文档无法回答”有两种可能一是检索确实没找到相关段落二是提示词太严格模型不敢回答。先看检索结果。如果检索出来的段落确实不相关那就是检索问题按上面的步骤排查。如果检索出来的段落相关但模型还是说不知道那就是提示词问题。可以试着把“如果文档中没有相关信息直接说不知道”改成“尽量根据文档回答如果信息不完整可以补充你的理解”看看模型是否愿意回答。我一般会保留严格的提示词因为宁可让模型说不知道也不要它编造答案。但如果发现模型过于保守可以适当放宽。5.3 回答太长或太短调整 max_tokens 和提示词回答长度不受控通常是max_tokens设得太大或太小。我一般设 800对于大多数问答够用。如果回答经常被截断调到 1200。如果回答太啰嗦可以在提示词里加一句“请简洁回答不超过 3 句话”。还有一个技巧在提示词里指定回答格式。比如“请用 bullet point 列出 3 个要点”这样模型会按格式输出长度自然受控。5.4 成本超预期从嵌入缓存和检索去重入手成本主要来自两块嵌入接口调用和模型生成。嵌入缓存能省掉重复文本的调用检索去重能减少送给模型的 token 数。去重的做法很简单检索出 top_k 个段落后计算两两之间的相似度如果超过 0.95只保留一个。这样能避免把几乎相同的段落都送给模型。def deduplicate_chunks(chunks, embeddings, threshold0.95): keep [] for i, emb in enumerate(embeddings): is_duplicate False for j in keep: if cosine_similarity(emb, embeddings[j]) threshold: is_duplicate True break if not is_duplicate: keep.append(i) return [chunks[i] for i in keep]这个函数在检索后调用能有效减少冗余。5.5 常见问题速查表问题现象可能原因排查方法解决方案检索结果不相关嵌入质量差算相似句子的余弦相似度换嵌入模型或清洗文本检索结果不相关切分太碎打印块文本调整切分策略增大块长度模型说不知道检索没召回检查 top_k 结果增大 top_k 或换嵌入模型模型说不知道提示词太严放宽提示词测试调整提示词措辞回答太长max_tokens 太大检查参数调小 max_tokens 或加格式要求回答太短max_tokens 太小检查参数调大 max_tokens成本超预期重复调用嵌入加日志统计加缓存和去重响应太慢检索或生成慢分步计时优化索引或换更快的模型6. 我踩过的坑和实测有效的经验6.1 别在切分上偷懒这是回报率最高的优化点我做过一个对比实验同一套文档同一套嵌入模型同一套提示词只改变切分策略。固定长度切分的 Recall5 是 0.72按段落切分是 0.81语义切分是 0.85。从 0.72 到 0.85提升非常明显而成本只是多写了几十行切分代码。所以我的建议是在切分上多花点时间比换更贵的模型划算得多。具体做法是先按段落切超长段落再按句子切句子之间保留重叠。如果效果还不够再考虑语义切分。6.2 嵌入模型不是越贵越好要看你的数据分布我试过用text-embedding-3-large替换text-embedding-3-small成本翻了 6 倍多但 Recall5 只提升了 0.02。对于我的数据中文技术文档这个提升完全不值。后来换成bge-large-zh本地部署Recall5 和text-embedding-3-small持平但成本几乎为零。所以选嵌入模型时一定要用自己的数据做评测。排行榜只能参考不能直接照搬。6.3 提示词里的“不知道”指令能省掉很多麻烦早期我做问答系统时没有加“如果不知道就说不知道”的指令结果模型经常编造答案。用户问一个文档里没有的问题模型也能编出一段看似合理的回答。后来加了这条指令幻觉率大幅下降。但要注意这条指令不能太生硬。我试过“如果文档中没有相关信息直接说不知道”模型有时候会过度保守明明有相关信息也说不知道。后来改成“只根据上述文档片段回答如果文档中没有相关信息直接说‘根据现有文档无法回答’”效果好很多。6.4 评估集要提前准备不要等上线了才想起来我见过太多项目开发阶段跑得飞快上线后用户反馈一堆问题但因为没有评估集根本不知道问题出在哪。我的做法是项目启动第一周就准备 20-30 个测试问题随着开发进展不断补充。评估集不需要很大但要有代表性。我一般会覆盖文档的主要章节每个章节出 2-3 个问题。问题类型要多样包括事实型、比较型、推理型。有了评估集每次改动都能快速验证效果避免盲目调参。6.5 日志要打全不然排查问题全靠猜AI 应用的链路很长从文档解析到最终回答中间经过多个模块。如果日志打不全出了问题根本不知道是哪一步的锅。我的做法是每个模块的输入输出都打日志关键参数也打。比如检索模块我会记录查询文本、查询嵌入向量前 10 维、top_k 结果、每个结果的相似度分数。生成模块我会记录提示词全文、模型参数、生成结果、token 消耗。这些日志在排查问题时非常有用。import logging logging.basicConfig(levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s) def answer_question_with_logging(query, chunks, index, top_k5): logging.info(fQuery: {query}) query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) logging.info(fTop-{top_k} indices: {indices[0]}) logging.info(fTop-{top_k} distances: {distances[0]}) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) logging.info(fPrompt length: {len(prompt)}) response client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800 ) answer response.choices[0].message.content logging.info(fAnswer: {answer}) logging.info(fToken usage: {response.usage}) return answer这套日志在开发阶段会输出很多信息但排查问题时能省下大量时间。生产环境可以调高日志级别只记录关键信息。6.6 本地嵌入模型部署省钱的代价是运维如果你决定用本地嵌入模型比如bge-large-zh需要额外考虑几件事模型下载、GPU 显存、推理速度、并发处理。我一般用sentence-transformers加载模型然后封装成一个批量推理函数from sentence_transformers import SentenceTransformer model SentenceTransformer(BAAI/bge-large-zh-v1.5) def get_embeddings_local(texts, batch_size32): embeddings model.encode(texts, batch_sizebatch_size, normalize_embeddingsTrue) return embeddings.tolist()normalize_embeddingsTrue会自动归一化省去了手动归一化的步骤。批量大小根据显存调整一般 32 或 64 比较合适。本地模型的优点是成本低、数据不出本地缺点是首次加载慢、需要 GPU、并发能力有限。如果你的数据量不大或者对成本敏感本地模型是不错的选择。6.7 重排序检索之后的第二道防线如果检索出来的 top_k 结果里相关段落排名靠后可以考虑加重排序模型。重排序的思路是先用嵌入检索出 top-20然后用一个更精细的模型对这 20 个结果重新打分取 top-5 送给大模型。常用的重排序模型有bge-reranker系列可以本地部署。重排序的代价是增加一次模型推理但能显著提升检索精度。我实测下来在 top-20 里加一层重排序Recall5 能提升 0.05-0.1。from sentence_transformers import CrossEncoder reranker CrossEncoder(BAAI/bge-reranker-large) def rerank(query, chunks, top_k5): pairs [[query, chunk] for chunk in chunks] scores reranker.predict(pairs) ranked_indices np.argsort(scores)[::-1][:top_k] return [chunks[i] for i in ranked_indices]重排序模型比嵌入模型大推理速度慢一些但只对少量候选做推理总体延迟增加不多。6.8 流式输出提升用户体验的简单手段大模型生成回答需要时间如果等全部生成完再返回用户会等很久。流式输出可以让用户看到回答逐字出现体验好很多。OpenAI 的接口支持流式输出def answer_question_stream(query, chunks, index, top_k5): query_emb np.array([get_embedding_cached(query)]).astype(float32) faiss.normalize_L2(query_emb) distances, indices index.search(query_emb, top_k) retrieved_chunks [chunks[i] for i in indices[0]] prompt build_prompt(query, retrieved_chunks) stream client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0.1, max_tokens800, streamTrue ) for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content用yield逐块返回前端可以实时渲染。这个改动很小但用户体验提升明显。6.9 多轮对话维护上下文的关键是会话状态单轮问答跑通后下一步是多轮对话。多轮对话的关键是维护会话状态用户问了什么、模型答了什么、检索了哪些文档。我一般用一个列表保存对话历史每次生成时把历史也拼进提示词。但要注意对话历史不能无限增长否则 token 会爆。我一般只保留最近 3-5 轮对话更早的对话做摘要或者直接丢弃。class Conversation: def __init__(self, max_history5): self.history [] self.max_history max_history def add_turn(self, role, content): self.history.append({role: role, content: content}) if len(self.history) self.max_history * 2: self.history self.history[-self.max_history * 2:] def get_context(self): return self.history多轮对话的检索策略也需要调整有时候需要结合历史问题来检索而不是只看当前问题。我一般会把最近一轮的用户问题和当前问题拼起来做检索效果比只看当前问题好。6.10 部署上线从脚本到服务的最后一公里开发阶段用脚本跑没问题上线需要封装成服务。我一般用 FastAPI 搭一个简单的 HTTP 接口from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): query: str top_k: int 5 app.post(/ask) def ask(request: QueryRequest): answer answer_question(request.query, chunks, index, request.top_k) return {answer: answer}然后用uvicorn启动uvicorn main:app --host 0.0.0.0 --port 8000这个服务很简单但足够支撑小规模使用。如果需要更高并发可以考虑加缓存、加队列、用异步接口。但那是另一个话题了。6.11 安全与合规别让模型说出不该说的话AI 应用上线前一定要考虑安全与合规。我一般会做几件事一是加输入过滤拦截明显恶意的查询二是加输出过滤检查模型回答是否包含敏感内容三是加日志审计记录所有查询和回答方便追溯。输入过滤可以用关键词匹配也可以用一个小模型做分类。输出过滤类似。这些措施不能保证 100% 安全但能挡住大部分明显问题。注意安全与合规是底线不要为了追求效果而忽略。上线前一定要过一遍安全审查确保没有明显漏洞。6.12 持续迭代上线只是开始AI 应用上线后需要持续迭代。我一般会关注几个指标用户查询量、检索命中率、回答满意度、平均响应时间、token 消耗。根据这些指标定期优化切分策略、嵌入模型、提示词模板。用户反馈是最宝贵的优化线索。我一般会在界面上加一个“这个回答有帮助吗”的按钮收集用户反馈。负面反馈的查询我会重点分析看看是检索问题还是生成问题。这套流程跑下来一个从零构建的 AI 工程化文档问答系统就完整了。它不依赖任何重型框架每个环节都清晰可控。你可以根据实际需求替换其中的任何模块比如把 FAISS 换成 Qdrant把 OpenAI 嵌入换成 bge把 GPT-4o-mini 换成其他模型。核心思路不变理解每个环节在干什么然后自己动手实现一遍。我在实际项目中发现这套从零构建的经验让我在使用成熟框架时更加得心应手。因为我知道框架在背后做了什么哪些地方可以信任哪些地方需要自己接管。这种掌控感是直接调 API 永远给不了的。