1. 项目拆解:AI工程从零到一到底在学什么
先聊点扎心的:现在网上聊 AI 开发的帖子,十个里有八个是“调 API 三分钟做个聊天机器人”。真到了生产环境,需求一复杂,数据一脏,并发一上来,那些 demo 全得推翻重写。我见过太多人 LangChain 用得飞起,RAG 跑起来也能出结果,但问一句“召回率怎么算”“chunk_size 为什么取 512”“上下文超长了你怎么办”,立马卡壳。
“ai-engineering-from-scratch”这个项目,核心思想全在这几个字里:不借助高级封装,从零开始把 AI 应用的每一层亲手搭一遍。
为什么值得这么干?理由很简单。框架封装度高,是把双刃剑。它能让你三天出 demo,也能让你三年看不懂底层。一旦线上出问题,黑盒的排查成本远超你当初省下的那点时间。写这个项目的目的,就是把你从“调用者”掰回“构建者”。从头写一遍调用逻辑、检索逻辑、Agent 循环、评估链路,你才知道每个组件到底在解决什么问题、性能瓶颈在哪、边界条件在哪。
适合谁来参考呢?三类人:
- 已经能用 LangChain/LlamaIndex 写 demo,但想更深一层理解原理的开发者;
- 正准备做 AI 应用落地,不想被框架绑架、想自己掌控全流程的工程师;
- 刚入门,想知道“AI 工程”到底包含哪些环节、应该怎么系统学习的新手。
这个仓库的理想学习路径,不是按框架文档走,而是按照 AI 应用的物理结构从薄到厚拆解:从大模型调用 → 上下文管理 → 检索增强 → 工具调用 → 评估反馈 → 部署监控。每加一层,你就更接近一个能扛住真实业务压力的系统。
2. 为什么“从零手写”比“直接上框架”更值得
我用个真实经历说明。之前做个内部知识库问答,最早图省事直接用 LangChain 的RetrievalQA,代码二十行,跑起来确实能答。结果上线第一天就翻车:用户问法稍有变化,检索回来的段落全是噪声,答非所问。我盯着 LangChain 的链式调用日志看了半天,不知道它内部做了哪些改写、用了什么检索策略,整个人懵住。
后来沉下心,把完整的 RAG 链路用原生代码重写了一遍,才发现问题出在 query 改写默认做了一堆花活,加上 top_k 取值过大,把无关段落一并塞进了上下文。两行参数的事,换成直接看底层代码,立刻一目了然。“从零手写”最大的价值,是让你拥有出问题时的定位能力。
再换个类比。你用自动挡开车,舒服是舒服,但车半路抛锚,你连引擎盖都不会开。从零手写就是逼你拆一遍发动机。虽然前期慢,但后面你对每个零件的脾气都门儿清。
从零手写还有几个实打实的优势:
- 灵活可控:生产环境的需求千奇百怪,框架预设的管线往往满足不了。自己搭,想怎么改怎么改。
- 便于排障:每一层都是你自己写的,日志、错误、性能瓶颈全在掌握之中。
- 规避兼容性地狱:LangChain 的版本更新有多频繁、Breaking Change 有多痛,经历过的人都懂。自己写,依赖少,长期稳定。
- 代码瘦身:调用一个几十兆的框架,只为用其中两个函数,冗余度高。裸写往往几千行代码就覆盖了核心业务。
当然,我不是劝你完全抛开 LangChain。等你亲手写过一遍底层逻辑,再回去用框架,你会看得懂它的设计、知道怎么定制。那时候框架才真正为你所用,而不是反过来。
3. 核心环节拆解:从模型到应用,每一层都在解决什么问题
3.1 模型层:打好地基,选对调用方式
AI 工程最底层的二选一:用托管 API,还是自部署模型。
托管 API 省心,但会碰到三座大山:成本不可控、数据出域、限流延迟。自部署模型用开源权重(Qwen、Llama、DeepSeek 等),前期折腾但长期成本低、可控性强。
工程上的建议是接口统一、模型可替换。所有代码不要跟某一家厂商的 SDK 强绑定,统一走 OpenAI 兼容协议。这样今天用 GPT-4o,明天换成 Qwen-Max,只改一个环境变量就行。这是我跟过的项目里最值得的一个决策——从没被单一厂商锁死过。
自部署的话,轻量场景用 Ollama 就够了,一行命令拉起模型,适合开发调试。生产环境追求吞吐,推荐 vLLM,PagedAttention 技术能把显存利用率拉高一大截,QPS 轻松翻倍。跑过 7B 模型的人应该深有体会,vLLM 与原生 transformers 部署的差距是肉眼可见的。
3.2 上下文工程:用一半费用做两倍效果
上下文管理是 AI 应用被低估的重头戏。很多工程问题,本质上就是上下文没管好。
- 裁剪:对话变长后,不是无穷加 token,而是把早期消息做摘要压缩。核心思路是“用更少的 token 保留关键信息”,窗口能塞下更多有效内容。
- 结构化注入:不要把整本书塞进 prompt,先用结构化提取,把关键事实压缩成条目,再喂给模型。
- 动态检索:不靠手工控制上下文,靠向量检索动态取最相关的片段。这就是 RAG 的雏形。
上下文管理的核心指标是“有效信息密度”。同样 8K 窗口,A 方案只能容纳 20 轮对话,B 方案能容纳 50 轮且关键信息不丢失,B 的架构明显更健康。这行当里,省钱和省事往往是一回事。
3.3 RAG 层:谁说加个向量库就叫 RAG
RAG 看起来简单:文档切割、向量化、检索、拼接。实际工程里,每个环节都有讲究。
切分文档是第一个暗坑。固定 500 字切一刀,会把一段完整语义拦腰截断。我踩过之后改成了语义分段:用段落 Markdown 标题、列表、空行做自然边界,必要时结合滑动窗口重叠。分块质量直接决定检索质量,这步偷懒,后面全白搭。
Embedding 模型选型也不能忽视。中文场景,bge-m3系列效果稳且开源可商用,text-embedding-3-large也很能打但按量付费、量大肉疼。实测下来,换成中文优化过的 embedding 模型,top-5 命中率能提升 15%-20%,你检索效果差的锅,模型至少背一半。
检索策略同样不能无脑 top-k。固定取前 5 段必然带噪声。更好的做法是:向量检索取前 20 → 重排模型(如bge-reranker)精排取前 3-4。步骤多了,精度立竿见影。重排是 RAG 效果从“能用”到“好用”的分水岭。
3.4 Agent 层:最小可行循环
Agent 核心不是概念,是那个循环:理解任务 → 选择合适的工具 → 调用并观察结果 → 决定下一步。工程实现上,看清每个环节:
- 意图识别:靠 prompt 明确列出可用工具和适用场景,模型负责路由。
- 工具调用:原生 OpenAI Function Calling 或开源模型自带的工具调用能力,把结构化参数传给函数。
- 循环控制:设最大迭代轮数(一般 5-8 轮)、条件终止、异常重试。不设上限,Agent 会自己死循环烧光你的预算,别问我是怎么知道的。
- 记忆组织:短期记忆存当前任务上下文,长期记忆靠外部存储(典型如向量库)。
自己从零实现 Agent 的价值在于:你能精准感知每一步消耗的时间、token 和成功率,而不是面对一个“智能体框架”干瞪眼。看日志从“天书”变成“正常人话”,瞬间舒服。
4. 从零实操:一步步搭建你自己的 AI 应用
这一节是硬货。我按“最小前向路径”走一遍,完整体验 AI 应用的诞生过程。环境默认 Python 3.10+。
4.1 第一步:搭一个最简 LLM 调用服务
先不碰任何框架,裸调模型 API。
# requirements.txt openai>=1.30.0 python-dotenv fastapi uvicorn# app.py import os from openai import OpenAI from dotenv import load_dotenv load_dotenv() _client = OpenAI( api_key=os.getenv("OPENAI_API_KEY"), base_url=os.getenv("OPENAI_BASE_URL") or "https://api.openai.com/v1" ) def chat(messages, model=None): model = model or os.getenv("LLM_MODEL", "gpt-4o-mini") response = _client.chat.completions.create( model=model, messages=messages, temperature=0.3, stream=True ) full_text = "" for chunk in response: delta = chunk.choices[0].delta.content if delta: full_text += delta return full_text几个关键点备注一下:
temperature=0.3:日常问答降到 0.3 能显著减少幻觉和发散。要创意写作再上调。- 流式接口:用户体验差异巨大。等 20 秒全量返回,用户早跑了;逐字输出,5 秒内就有“在干活”的感觉。生产环境务必流式。
- 统一 base_url:换模型厂商零改动。我在
.env里放OPENAI_BASE_URL,切换供应商时只改配置。
4.2 第二步:手写一个 RAG 全链路
这是整个项目的核心部分。完整实现下来的记忆点异常清晰,比看十篇教程管用。
# rag.py from openai import OpenAI import sqlite3 class VectorStore: """零依赖向量库:把向量存 SQLite,检索时全表扫""" def __init__(self, db_path): self.conn = sqlite3.connect(db_path) self.conn.execute("CREATE TABLE IF NOT EXISTS vectors (id TEXT PRIMARY KEY, embedding TEXT, content TEXT)") self.client = OpenAI() def embed(self, text): resp = self.client.embeddings.create( model="text-embedding-3-small", input=text ) return resp.data[0].embedding def add(self, doc_id, content): vec = self.embed(content) self.conn.execute( "INSERT INTO vectors VALUES (?, ?, ?)", (doc_id, json.dumps(vec), content) ) self.conn.commit() def search(self, query, top_k=5): q_vec = self.embed(query) rows = self.conn.execute("SELECT * FROM vectors").fetchall() scored = [] for doc_id, emb_json, content in rows: emb = json.loads(emb_json) # 余弦相似度 dot = sum(a*b for a, b in zip(q_vec, emb)) norm_q = sum(a*a for a in q_vec) ** 0.5 norm_e = sum(b*b for b in emb) ** 0.5 score = dot / (norm_q * norm_e + 1e-9) scored.append((score, content)) scored.sort(reverse=True, key=lambda x: x[0]) return scored[:top_k]思路很直白:全表扫描 + 余弦相似度。性能远不如专业向量库,但作为教学训练,逻辑一目了然,便于理解核心机制。生产环境记得换 Milvus 这类专用引擎。
实际文档切分我用的是递归字符切分器,附上核心逻辑:
def chunk_text(text, chunk_size=512, overlap=80): """按分隔符优先策略切分,保留上下文重叠""" separators = ["\n\n", "\n", "。", "!", "?", ";"] current = "" for char in text: current += char if len(current) >= chunk_size: # 找最近的边界符切开 for sep in separators: idx = current.rfind(sep) if idx != -1: yield current[:idx+len(sep)] current = current[idx+len(sep):] break else: yield current current = "" if current: yield current切分参数不是拍脑袋。512 是中等长度平衡点:太长,embedding 平均了语义导致检索不精准;太短,上下文碎片化丢失逻辑。80 的重叠让相邻片段保持线索。文本结构化越强的文档,切分效果越依赖分隔符匹配,这是我实测多轮之后的结论。
最后“检索 + 生成”拼装代码如下:
def rag_query(question): chunks = store.search(question, top_k=5) context = "\n\n".join(chunk for _, chunk in chunks) messages = [ {"role": "system", "content": f"基于以下资料回答问题。资料不充分时,直接说不知道,不要编造。\n\n资料:\n{context}"}, {"role": "user", "content": question} ] return chat(messages)这套组合足以支撑起一个企业内部知识库问答应用。生产化还要加上:文档更新时自动重算 embedding、切分策略按文档类型配置、检索结果缓存等。但骨架就是这样。
4.3 第三步:给 Agent 加上工具调用
从零实现 Agent,绕不开工具调用。我用最简单的天气查询示例展示设计模式:
import json TOOLS = [ { "type": "function", "function": { "name": "get_weather", "description": "查询城市当前天气", "parameters": { "type": "object", "properties": { "city": {"type": "string", "description": "城市名"} }, "required": ["city"] } } } ] def get_weather(city): # 这里接真实天气 API return f"{city}:多云,24℃,东风3级" def agent_run(user_input, max_steps=5): messages = [{"role": "user", "content": user_input}] for step in range(max_steps): resp = client.chat.completions.create( model="gpt-4o-mini", messages=messages, tools=TOOLS, tool_choice="auto" ) msg = resp.choices[0].message if not msg.tool_calls: return msg.content messages.append(msg) for tool_call in msg.tool_calls: args = json.loads(tool_call.function.arguments) result = globals()[tool_call.function.name](**args) messages.append({ "role": "tool", "tool_call_id": tool_call.id, "content": result }) return "达到最大步数,已停止。"这个模式就是所有 Agent 框架的底层骨架。max_steps=5为了防止模型陷入死循环。我实测过,不设上限的 Agent 经常在工具调用里转圈出不来,token 烧得人心疼。这也是生产环境永不省略的保险丝。
4.4 第四步:评估和回归——调优的照妖镜
没有评估体系的 AI 应用优化,全凭运气。自己手写 RAG 时,一定要把评估搭起来:
EVAL_SET = [ {"question": "公司的年假制度是怎样的?", "expected_keywords": ["累计工作年限", "年假天数"]}, {"question": "报销流程需要几步?", "expected_keywords": ["提单", "审批", "财务打款"]}, # ... 大概50条以上,覆盖各业务场景 ] def evaluate(): total, hit = 0, 0 for item in EVAL_SET: answer = rag_query(item["question"]) total += 1 if all(kw in answer for kw in item["expected_keywords"]): hit += 1 return hit / total这个简单精确的“关键词命中率”指标,能当回归测试的守门员。每次改切分策略、换 embedding、调 top_k,跑一遍看分数变化——涨了就保留,跌了就打回。
更精细一点,可以用LLM-as-judge:
def judge_answer(question, answer, reference): prompt = f"""你是评估助手。判断以下回答是否准确。 问题:{question} 参考答案:{reference} 模型回答:{answer} 输出:完全正确 / 部分正确 / 错误""" return chat([...]) # 返回上述三选一实践下来,先埋头搭好了评估集再做优化,效率是高一个层级的。没有度量,一切调优都等于蒙眼开车。
4.5 第五步:部署上线与监控
工程化最后一步是让应用真正 7x24 跑起来。上 FastAPI 封装:
from fastapi import FastAPI from pydantic import BaseModel app = FastAPI() class Query(BaseModel): question: str @app.post("/query") def query_endpoint(q: Query): answer = rag_query(q.question) return {"answer": answer}服务化暴露 HTTP 接口,前端/后端系统都能直接调用。别用 Jupyter Notebook 当生产服务,掉线一次口碑崩一次。
部署容器化 Dockerfile 要点:
FROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]重头戏在监控。我强烈建议每个 AI 工程在日志之外加一层可观测性埋点,至少包含:
- Token 消耗和成本:每轮对话花了多少钱,按用户聚合统计;
- 响应延迟分段:embedding 耗时、检索耗时、生成耗时分开记录,定位瓶颈一目了然;
- 检索命中率:用户点没点“有帮助”,是最直接的反馈信号;
- 失败率:4xx/5xx、超时、拒绝率。
实测过 Langfuse,埋点完成后 dashboard 能直接显示每次调用的完整轨迹,花钱花在哪清清楚楚,排查问题效率极高。
5. 实战避坑:我在从零构建中踩过的真实问题
跟完上述步骤,你大概率会遇到下面这些坎,我提前把雷给你排了。
5.1 上下文吞金兽:prompt 越堆越多
最典型的问题。对话超过 20 轮,上下文里塞满了历史,每次调用把老账都翻出来算钱,响应也慢得像蜗牛。
排查思路:先打印每次 request 的 token 数,把缺失的信息暴露出来。解决三板斧:早期消息摘要化(详情见 3.2);窗口长度限制+滑动截断;对用户输入做长度预检,超额先压缩再入模型。
5.2 检索命中率虚低:问题出在切分,不是模型
有段时间我把切分调成 256 字后,检索全乱套。查了半天才发现固定长度切分把语义拆碎了。比如“报销流程”四个字段落在前一片,详细步骤说明在后一片,检索到的片段全是残肢。
解决思路:切分策略与文档结构强相关。Markdown 标题明确的,先按标题切;法律/说明书类句子自成段的,按句子切。确定策略后在评估集上跑分对比,别用感觉选方案。
5.3 模型乱编答案:幻觉压不下去
幻觉的根源,是模型在“知识空白区”做了补全。我的处理矩阵:
- 系统提示里写死“资料里没写就回答不知道,禁止自行补充”;
temperature从 0.7 压到 0.2,效果很明显;- 要求输出引用来源,答案自带可追溯性;
- 对关键数据问题走“检索-抽取-比对”逻辑,让模型输出结论后必须附上原文片段。
5.4 线上响应慢:卡在生成环节
AI 应用的延迟大头几乎都在生成阶段。RAG 链路里 embedding 加检索大概占 300ms,模型生成 500 token 要 6-10 秒。
优化手段:流式输出先让用户动起来;小问题用小模型路由(比如简单问答走 4o-mini,复杂逻辑走大模型);对高频问题做 exact-match 缓存;需要重逻辑的场景加前缀缓存。
5.5 第三方 API 抖动:重试策略直接决定体验
线上跑了一段时间你就会发现,再好的 API 也会偶尔抽风。工程上要实装:指数退避重试(1s、2s、4s,最大三次)、超时设置(connect 10s / read 60s)、失败降级策略(主模型故障切备选模型),这些就是应用稳定性的压舱石。
6. 从零到产线:下一个升级目标是哪三个方向
“from-scratch”项目跑通之后,整个体系的升级路线一般有三个优先方向,按投入产出比排优先级:
第一优先级:数据飞轮。把线上用户真实提问采集下来,每周筛选 badcase,补充进评估集。做评估不是为了这一版,是为了下一版。跑得越久,评估集越厚,系统调优的目标越清晰。
第二优先级:多路召回融合。单路向量检索能力有上限,试着叠加 BM25 关键词检索、SQL 结构化查询,最后用重排模型融合。QA 准确率通常能再上 5-10 个百分点。
第三优先级:领域微调(若确实需要)。RAG 跑通后,如果某些专业术语的生成风格始终不对,考虑用 LoRA 做一次轻量微调。这步是加分项,前两个方向没做扎实的情况下不建议碰。
实操中的心得体会,永远记住一句话:先跑通,再调优,先动手,再换框架。只要核心链路是你亲手搭的,后续无论接 LangGraph、上 Agent 工作流、做多模态,都不会慌。地基是你打的话,盖楼心里就有底了。