☰
后端老兵从零搭建AI工程能力:踩坑复盘与RAG实战指南
2026/10/3 14:33:22 网站建设 项目流程

1. 从零搭建AI工程能力:一个后端老兵的踩坑与复盘

"ai-engineering-from-scratch"这个标题,第一次看到的时候我愣了一下。不是因为陌生,恰恰相反——过去两年里,我身边至少有七八个后端、前端甚至运维的朋友问过我类似的问题:"我想转AI工程,但不知道从哪下手,是不是得先把线性代数重新学一遍?"

这个项目标题背后要解决的核心问题其实很明确:一个没有机器学习背景的普通开发者,如何从零开始构建一套能真正落地干活的AI工程能力体系。注意,是AI工程,不是AI研究。这两者之间的差别,比很多人想象的要大得多。

AI工程的核心不是推导反向传播公式,也不是手撸Transformer。它的核心是:把已有的模型能力,通过工程手段,稳定、高效、可维护地交付到业务场景中去。这包括数据处理管线的搭建、模型服务的部署与扩缩容、推理性能的优化、提示词的管理与版本控制、评测体系的建立、成本控制等等。这些事情,一个有过后端开发经验的人,其实上手比纯算法背景的人还要快。

这篇文章适合谁看?如果你是后端、全栈、数据工程师,想系统性地补齐AI工程能力;或者你是刚入行的算法工程师,发现学校教的东西和实际工作差距很大;再或者你是技术负责人,想给团队规划一条从零到一的AI工程学习路径——那这篇内容应该能给你一些直接能用的参考。

我自己的背景是后端开发,做了六年微服务架构,两年前开始系统性地接触AI工程。踩过的坑不算少,从环境配置到模型部署,从提示词管理到线上推理的延迟优化,几乎每个环节都交过学费。下面我把整个从零搭建的过程拆开来讲,尽量把每一步的"为什么"说清楚,让你少走弯路。

2. 整体学习路径设计与技术选型思路

2.1 为什么不能按传统ML课程的路子走

市面上大部分AI学习资源,走的是"机器学习导论→深度学习→NLP/CV→大模型"这条线。这条路本身没问题,但它的目标函数是培养研究者,不是工程师。你会花大量时间在数学推导、论文复现、模型结构创新上,而这些在实际AI工程工作中占比可能不到10%。

我试过按这条路走了三个月,学完了吴恩达的机器学习课程,啃了一半《深度学习》花书,结果发现:面对一个"把公司客服系统接入大模型"的需求,我依然不知道从哪下手。因为真正的工程问题——怎么管理API密钥、怎么处理超时重试、怎么设计提示词模板、怎么评估输出质量、怎么控制token成本——这些课程里根本不讲。

所以我的建议是:如果你目标是AI工程,就不要从数学开始,从"跑通一个最小可用系统"开始。先建立工程直觉,再按需补理论。这跟学Web开发是一个道理——你不会先学TCP/IP协议栈再写第一个HTML页面。

2.2 分阶段的能力建设路线

我把整个学习路径分成四个阶段,每个阶段都有明确的产出物,避免"学了很多但什么都做不出来"的困境。

第一阶段:基础环境与API调用能力(1-2周)

这个阶段的目标很简单:能熟练调用主流大模型的API,理解请求/响应的基本结构,能处理常见的错误情况。产出物是一个命令行小工具,比如一个能读取本地文档并回答问题的脚本。

第二阶段:应用开发框架与RAG基础(3-4周)

掌握至少一个AI应用开发框架,理解检索增强生成(RAG)的基本原理和实现方式。产出物是一个能基于私有知识库问答的Web应用。

第三阶段:工程化与生产部署(4-6周)

这个阶段是区分"玩具项目"和"生产系统"的关键。你需要掌握:提示词版本管理、输出质量评测、推理性能优化、成本监控、日志与可观测性。产出物是一个有完整监控和评测体系的AI服务。

第四阶段:进阶与专项深入(持续)

根据你的业务方向选择深入:模型微调、Agent系统、多模态应用、推理加速等。

2.3 技术选型的几个关键决策

在工具选型上,我踩过一些坑,这里直接给结论和理由。

编程语言:Python为主,但不要放弃你原有的语言

Python是AI工程的事实标准,生态最完善。但如果你原本是Java或Go背景,不要觉得需要完全转语言。很多生产系统的架构是:Python做模型推理服务,Java/Go做业务编排层。我现在的做法就是Python写推理和数据处理,Go写API网关和业务逻辑。

开发框架:LangChain vs LlamaIndex vs 裸写

我的建议是:先用裸写,再用框架。裸写的意思是直接用OpenAI SDK或类似的库,自己管理提示词、上下文、工具调用。这样你能真正理解每一步在发生什么。等你把RAG的每个环节都手写过一遍之后,再用LangChain或LlamaIndex去提效,这时候你就能判断哪些抽象是合理的,哪些是过度封装。

我见过太多人一上来就用LangChain,结果出了问题完全不知道从哪排查,因为框架把太多细节藏起来了。

向量数据库:从最简单的开始

新手最容易犯的错是一上来就上Milvus或Weaviate这种重型向量数据库。实际上,如果你的数据量在百万级以下,用FAISS或者甚至PostgreSQL的pgvector扩展就完全够了。我第一个RAG项目用的是Chroma,本地文件存储,零配置,跑起来再说。等数据量真的上来了再迁移,迁移成本远比你想象的低。

模型选择:不要一上来就追求最强模型

GPT-4级别的模型确实强,但成本也高。实际工程中,大量任务用中小模型就能做好。我的策略是:先用最强模型跑通流程、建立评测基线,然后逐步降级到更便宜的模型,看效果损失是否可接受。很多时候,一个精心设计的提示词加上小模型,效果比粗糙提示词加大模型还好。

3. 核心环节拆解与实操要点

3.1 环境搭建:比你想象的简单,但有几个坑

AI工程的环境搭建其实比传统后端开发简单,因为大部分工作是通过API完成的,不需要本地GPU。但有几个坑我必须提前说。

Python环境管理

不要用系统自带的Python。用pyenv或conda管理多版本,用venv或poetry管理项目依赖。我推荐pyenv + poetry的组合,前者管Python版本,后者管包依赖。

# 安装pyenv(macOS/Linux) curl https://pyenv.run | bash # 安装指定Python版本 pyenv install 3.11.6 pyenv global 3.11.6 # 项目初始化 mkdir ai-engineering-lab && cd ai-engineering-lab poetry init poetry add openai tiktoken python-dotenv

为什么强调版本管理?因为AI领域的库更新极快,不同项目依赖的版本经常冲突。我有个项目因为transformers版本不对,排查了整整一个下午。

API密钥管理

这是新手最容易忽视的安全问题。绝对不要把API密钥硬编码在代码里,也不要提交到Git仓库。用.env文件加python-dotenv管理,并且把.env加入.gitignore。

# config.py import os from dotenv import load_dotenv load_dotenv() OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") if not OPENAI_API_KEY: raise ValueError("OPENAI_API_KEY not set in environment")

注意:即使是在个人项目里,也要养成密钥管理的习惯。我见过有人把密钥推到公开仓库,几分钟内就被扫到并盗用,账单直接飙到几百美元。

网络与代理配置

调用海外模型API时,网络稳定性是个现实问题。我的做法是在代码层面做好超时和重试,而不是依赖网络层面的方案。

from openai import OpenAI import httpx client = OpenAI( api_key=OPENAI_API_KEY, timeout=httpx.Timeout(30.0, connect=5.0), max_retries=3 )

超时设置很关键。连接超时设短一点(5秒),读取超时设长一点(30秒),因为大模型生成响应本身就需要时间。重试次数设3次,配合指数退避。

3.2 提示词工程:不是玄学,是工程

提示词工程经常被神秘化,好像是什么独门秘籍。其实它本质上就是用自然语言做编程,有章法可循。

结构化提示词模板

我习惯把提示词拆成几个固定部分:角色定义、任务描述、输入数据、输出格式、约束条件。这样便于管理和复用。

PROMPT_TEMPLATE = """你是一个{role}。 ## 任务 {task} ## 输入 {input_data} ## 输出格式 {output_format} ## 约束 {constraints} """

为什么要这么拆?因为当输出效果不好时,你能快速定位是哪个部分的问题。是角色定义不清晰?还是输出格式没约束好?如果提示词是一大段散文,排查起来就很痛苦。

Few-shot示例的选择

给示例不是越多越好。我的经验是:2-3个高质量示例,覆盖边界情况,比10个相似示例效果好。而且示例的顺序有影响,把最典型的放在最后,因为模型对靠近生成位置的上下文更敏感。

提示词的版本管理

这是很多个人项目忽视的环节。提示词应该像代码一样管理:存在文件里,用Git追踪,每次修改记录原因和效果变化。

prompts/ qa_system/ v1_basic.txt v2_with_citation.txt v3_strict_format.txt CHANGELOG.md

CHANGELOG里记录每次修改的动机和评测结果。这样当线上效果波动时,你能快速回滚到之前的版本。

3.3 RAG系统的核心:检索质量决定一切

RAG(检索增强生成)是当前AI工程最核心的应用模式。但很多人把精力花在生成端,忽视了检索端。实际上,RAG系统效果的上限由检索质量决定。

文档切分策略

切分不是简单地按固定字数切。我试过几种策略,效果最好的是按语义边界切分,配合重叠窗口。

from langchain.text_splitter import RecursiveCharacterTextSplitter splitter = RecursiveCharacterTextSplitter( chunk_size=500, chunk_overlap=100, separators=["\n\n", "\n", "。", "!", "?", ".", " ", ""] )

chunk_size设500个字符左右,overlap设100。为什么?太小了语义不完整,太大了检索精度下降。overlap是为了避免关键信息刚好被切在边界上。

Embedding模型的选择

不要盲目追求最大的embedding模型。我对比过几个常用模型,在中文场景下,BGE系列和M3E系列性价比很高。关键是:embedding模型必须和你的业务语料匹配。通用模型在专业领域(比如医疗、法律)效果会打折扣。

检索策略:向量检索不够,要加关键词检索

纯向量检索有个问题:对精确匹配不敏感。比如用户搜"GPT-4的上下文窗口是多少",向量检索可能返回一堆关于"大模型上下文"的泛泛内容,而漏掉精确提到"GPT-4"和"128k"的文档。

我的做法是混合检索:向量检索 + BM25关键词检索,然后用RRF(倒数排名融合)合并结果。

def hybrid_search(query, vector_store, bm25_index, top_k=5): vector_results = vector_store.search(query, top_k=top_k*2) bm25_results = bm25_index.search(query, top_k=top_k*2) # RRF融合 scores = {} for rank, doc in enumerate(vector_results): scores[doc.id] = scores.get(doc.id, 0) + 1/(60 + rank) for rank, doc in enumerate(bm25_results): scores[doc.id] = scores.get(doc.id, 0) + 1/(60 + rank) sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True) return [doc_id for doc_id, _ in sorted_docs[:top_k]]

这个RRF的常数60是原论文推荐的,实测下来确实比简单加权求和稳定。

重排序(Rerank)

检索出候选文档后,用一个交叉编码器做重排序,能显著提升精度。我常用的是BGE-reranker,部署简单,效果提升明显。代价是增加一点延迟,但在大多数场景下值得。

4. 完整实操流程:从零搭建一个可用的RAG问答系统

4.1 项目结构与依赖

先看整体结构,这样你心里有个全局图。

rag-qa-system/ ├── config/ │ ├── settings.py │ └── prompts/ │ └── qa_v1.txt ├── data/ │ ├── raw/ # 原始文档 │ └── processed/ # 处理后的chunk ├── src/ │ ├── ingestion/ # 文档摄入 │ │ ├── loader.py │ │ ├── splitter.py │ │ └── embedder.py │ ├── retrieval/ # 检索 │ │ ├── vector_store.py │ │ ├── bm25.py │ │ └── hybrid.py │ ├── generation/ # 生成 │ │ └── qa_chain.py │ └── evaluation/ # 评测 │ └── evaluator.py ├── tests/ ├── pyproject.toml └── README.md

依赖清单:

[tool.poetry.dependencies] python = "^3.11" openai = "^1.10.0" chromadb = "^0.4.22" rank-bm25 = "^0.2.2" sentence-transformers = "^2.3.0" pypdf = "^4.0.0" python-dotenv = "^1.0.0" tiktoken = "^0.6.0"

4.2 文档摄入管线

文档摄入分三步:加载、切分、向量化。

# src/ingestion/loader.py from pathlib import Path from pypdf import PdfReader def load_documents(data_dir: str) -> list[dict]: docs = [] for path in Path(data_dir).rglob("*"): if path.suffix == ".pdf": reader = PdfReader(path) text = "\n".join(page.extract_text() for page in reader.pages) docs.append({"source": str(path), "text": text}) elif path.suffix in (".md", ".txt"): text = path.read_text(encoding="utf-8") docs.append({"source": str(path), "text": text}) return docs

切分时保留元数据很重要,这样生成回答时能标注来源。

# src/ingestion/splitter.py from langchain.text_splitter import RecursiveCharacterTextSplitter def split_documents(docs: list[dict], chunk_size=500, overlap=100): splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=overlap, separators=["\n\n", "\n", "。", "!", "?", ".", " ", ""] ) chunks = [] for doc in docs: for i, chunk_text in enumerate(splitter.split_text(doc["text"])): chunks.append({ "id": f"{doc['source']}::{i}", "text": chunk_text, "source": doc["source"], "chunk_index": i }) return chunks

向量化用sentence-transformers,本地跑,不花钱。

# src/ingestion/embedder.py from sentence_transformers import SentenceTransformer class Embedder: def __init__(self, model_name="BAAI/bge-small-zh-v1.5"): self.model = SentenceTransformer(model_name) def embed(self, texts: list[str]) -> list[list[float]]: # BGE模型建议加instruction前缀 prefixed = [f"为这个句子生成表示以用于检索:{t}" for t in texts] return self.model.encode(prefixed, normalize_embeddings=True).tolist()

注意:BGE系列模型在检索任务上,query和document的处理方式不同。query需要加instruction前缀,document不需要。这个细节很多人忽略,导致检索效果差一截。

4.3 混合检索实现

向量库用Chroma,BM25用rank-bm25库。

# src/retrieval/vector_store.py import chromadb class VectorStore: def __init__(self, persist_dir="./chroma_db"): self.client = chromadb.PersistentClient(path=persist_dir) self.collection = self.client.get_or_create_collection( name="documents", metadata={"hnsw:space": "cosine"} ) def add(self, chunks, embeddings): self.collection.add( ids=[c["id"] for c in chunks], documents=[c["text"] for c in chunks], metadatas=[{"source": c["source"]} for c in chunks], embeddings=embeddings ) def search(self, query_embedding, top_k=10): results = self.collection.query( query_embeddings=[query_embedding], n_results=top_k ) return list(zip(results["ids"][0], results["documents"][0]))

BM25索引:

# src/retrieval/bm25.py from rank_bm25 import BM25Okapi import jieba class BM25Index: def __init__(self, chunks): self.chunks = chunks self.tokenized = [list(jieba.cut(c["text"])) for c in chunks] self.bm25 = BM25Okapi(self.tokenized) def search(self, query, top_k=10): tokens = list(jieba.cut(query)) scores = self.bm25.get_scores(tokens) top_indices = sorted(range(len(scores)), key=lambda i: scores[i], reverse=True)[:top_k] return [(self.chunks[i]["id"], self.chunks[i]["text"]) for i in top_indices]

中文分词用jieba,英文场景可以用简单的空格切分。BM25对中文的效果,分词质量影响很大,这是很多人踩的坑。

4.4 生成与引用标注

生成环节的关键是强制模型基于检索到的内容回答,并标注来源。

# src/generation/qa_chain.py from openai import OpenAI QA_PROMPT = """你是一个严谨的问答助手。请仅基于以下参考资料回答问题。 ## 参考资料 {context} ## 问题 {question} ## 要求 1. 如果参考资料中没有相关信息,直接回答"根据现有资料无法回答该问题" 2. 回答时用[1][2]标注引用的资料编号 3. 不要编造任何资料中没有的信息 ## 回答 """ class QAChain: def __init__(self, client, retriever): self.client = client self.retriever = retriever def answer(self, question: str) -> dict: docs = self.retriever.search(question, top_k=5) context = "\n\n".join( f"[{i+1}] {text}" for i, (_, text) in enumerate(docs) ) prompt = QA_PROMPT.format(context=context, question=question) response = self.client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.1 ) return { "answer": response.choices[0].message.content, "sources": [doc_id for doc_id, _ in docs], "usage": response.usage.total_tokens }

temperature设0.1,因为问答任务需要稳定和准确,不需要创造性。这个参数很多人设太高,导致同一个问题每次回答都不一样。

4.5 评测体系:没有评测就没有优化

这是最容易被忽视但最重要的环节。你需要一套自动化的评测流程,否则每次改提示词或换模型都是盲猜。

# src/evaluation/evaluator.py class RAGEvaluator: def __init__(self, qa_chain, test_cases): self.qa_chain = qa_chain self.test_cases = test_cases # [{"question": ..., "expected_keywords": [...]}] def evaluate(self): results = [] for case in self.test_cases: output = self.qa_chain.answer(case["question"]) # 关键词命中率 hit = sum( 1 for kw in case["expected_keywords"] if kw in output["answer"] ) / len(case["expected_keywords"]) results.append({ "question": case["question"], "answer": output["answer"], "keyword_hit_rate": hit, "tokens": output["usage"] }) return results

测试用例至少准备20-30条,覆盖:事实性问题、需要多文档综合的问题、资料中没有答案的问题(测试拒答能力)、边界问题。每次修改系统后跑一遍,看指标变化。

5. 常见问题与排查技巧实录

5.1 检索相关的问题

问题:检索到的文档和问题不相关

排查顺序:先看embedding模型是否适合你的语料。如果是专业领域,通用embedding模型效果会差。其次看chunk_size是否合适,太大导致语义稀释,太小导致信息不完整。最后检查query是否加了正确的instruction前缀。

问题:明明文档里有答案,但检索不到

这通常是关键词匹配的问题。纯向量检索对精确术语不敏感。解决方案是加BM25混合检索。另外检查文档切分时是否把关键信息切断了,适当增加overlap。

问题:检索结果重复度高

多个chunk来自同一文档的相邻位置,内容高度相似。解决方案是在检索后做去重,或者用MMR(最大边际相关性)算法选择多样性结果。

5.2 生成相关的问题

问题:模型编造资料中没有的信息

这是RAG最常见的失败模式。解决方案:提示词里明确约束"仅基于参考资料回答",temperature调低,加few-shot示例展示正确的拒答行为。如果还是不行,考虑在生成后加一个验证步骤,检查回答中的每个事实是否能在检索文档中找到依据。

问题:回答格式不稳定

用结构化输出。OpenAI的API支持response_format参数指定JSON schema,或者用function calling强制格式。不要指望提示词能100%约束住格式。

问题:长文档处理时超出上下文窗口

不要把所有检索结果都塞进上下文。按相关性排序,取top_k个,并且对每个chunk做长度限制。如果单个chunk太长,做二次摘要。

5.3 性能与成本问题

问题:响应太慢

瓶颈通常在embedding计算和LLM生成。embedding可以本地GPU加速或批量处理。LLM生成可以换更快的模型,或者用流式输出改善用户体验(首token时间比总时间更重要)。

问题:成本失控

监控token消耗,设置预算告警。优化策略:用更小的模型处理简单问题(路由策略),缓存常见问题的回答,压缩提示词减少输入token。

问题类型排查方向快速验证方法
检索不相关embedding模型、chunk策略手动检查top_k结果
检索漏召回关键词匹配、overlap加BM25对比效果
生成编造提示词约束、temperature构造无答案测试用例
格式不稳定结构化输出检查API参数
响应慢模型选择、流式输出分段计时
成本高token监控、模型路由统计每次调用token

5.4 几个我踩过的坑

坑一:忽视文档预处理质量

PDF提取的文本经常有乱码、断行、页眉页脚。直接拿来做embedding,效果很差。必须做清洗:去除页眉页脚、合并断行、处理特殊字符。我有个项目因为PDF提取质量差,检索效果一直上不去,后来花了一天做文本清洗,效果提升明显。

坑二:embedding模型和生成模型不匹配

有些embedding模型是在英文语料上训练的,用在中文场景效果打折。选embedding模型时,一定要在你的实际语料上测试检索效果,不要只看排行榜。

坑三:没有做query改写

用户的原始问题往往口语化、有歧义。加一步query改写,把用户问题转成更适合检索的形式,能显著提升召回率。比如用户问"那个新出的模型多少钱",改写成"最新模型API定价"再检索。

坑四:忽视缓存

很多问题是重复的。加一层语义缓存,相似问题直接返回缓存结果,能省大量token。用embedding做相似度匹配,阈值设0.95以上比较安全。

6. 从能用走向好用:工程化进阶要点

6.1 可观测性建设

生产系统和玩具项目的区别,很大程度在于可观测性。你需要知道:每次请求的检索结果是什么、生成了什么、消耗了多少token、延迟分布如何。

import logging import time logger = logging.getLogger("rag") def traced_answer(chain, question): start = time.time() result = chain.answer(question) latency = time.time() - start logger.info({ "question": question, "retrieved_ids": result["sources"], "answer_length": len(result["answer"]), "tokens": result["usage"], "latency_ms": latency * 1000 }) return result

这些日志积累起来,就是你优化的依据。哪些问题检索效果差、哪些问题token消耗高,一目了然。

6.2 提示词与配置的版本管理

把提示词、模型参数、检索参数都抽到配置文件里,用Git管理。每次变更记录效果指标。这样当效果回退时,能快速定位是哪个变更导致的。

# config/qa_config.yaml version: "1.2.0" model: name: "gpt-4o-mini" temperature: 0.1 max_tokens: 1000 retrieval: top_k: 5 use_rerank: true hybrid_alpha: 0.5 prompt: file: "prompts/qa_v3.txt"

6.3 持续评测与回归测试

每次修改系统后,自动跑评测集,对比关键指标。指标下降超过阈值就告警。这套流程建立起来后,你就能放心地迭代优化,不用担心改坏。

评测指标建议关注:检索命中率(Recall@k)、回答准确率、拒答准确率、平均延迟、平均token消耗。前三个衡量质量,后两个衡量成本和性能。

6.4 安全与合规考量

生产系统必须考虑:输入过滤(防止提示词注入)、输出过滤(防止敏感信息泄露)、访问控制、审计日志。提示词注入是当前很现实的风险,用户可能在输入里嵌入指令试图操控模型行为。基本的防护是在系统提示词里明确边界,并对用户输入做检测。

我在实际项目中的体会是,AI工程最难的不是某个技术点,而是建立一套可持续迭代的工程体系。模型会更新,提示词会调整,业务需求会变化,只有把评测、监控、版本管理这些基础设施建好,你才能快速响应变化而不失控。从零开始的时候,不要追求一步到位,先跑通最小闭环,再逐步加固。每加一个环节,都要问自己:这个环节解决什么问题,没有它行不行。这样能避免过度工程,也能确保每一步都是必要的。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询