从零手搓AI工程：推理、RAG与Agent全链路实战

1. 从零手搓AI工程：为什么我不建议你直接调包

很多人一上来就想搞个大模型应用，第一反应是找个API接上，或者拉个开源框架跑个demo。结果呢？demo跑通了，一上生产就崩。延迟高得离谱，成本失控，稍微改个需求就发现整个架构得推倒重来。我见过太多团队在这个阶段反复横跳，最后项目黄了，人也没成长。

ai-engineering-from-scratch这个标题，核心不是让你从零训练一个GPT，那玩意儿不是个人能玩的。它的真正价值在于：把AI工程拆解成可理解、可动手、可调试的原子单元。你要亲手实现一个最小的推理引擎、一个最简单的RAG流程、一个最朴素的Agent循环。只有自己写过一遍，才知道框架帮你藏了多少坑，也才知道什么时候该用框架，什么时候该自己写。

这篇文章适合谁？如果你是刚转AI方向的后端工程师，或者想从算法调参侠转型为AI系统工程师，再或者你是个技术负责人想搞清楚AI应用到底该怎么架构，那这篇内容就是给你写的。我会把整个从零构建AI工程能力的路径拆成四个阶段：推理基础、检索增强、Agent编排、性能与成本。每个阶段都给你可运行的代码骨架和踩坑记录。

注意：本文所有代码示例基于Python生态，但思路不绑定任何特定框架。你完全可以用Go或Rust复现同样的逻辑。

2. 推理基础：手写一个能用的推理循环

2.1 为什么必须自己实现一次推理循环

你可能会问，transformers库一行代码就能推理，为什么要自己写？因为你不知道那一行背后发生了什么。当你遇到显存溢出、生成结果乱码、流式输出卡顿的时候，你连从哪查起都不知道。

自己实现一次推理循环，你会被迫理解这几个核心概念：tokenization、KV Cache、采样策略、停止条件。这四个东西搞明白了，后面所有优化都有根基。

先看一个最简化的推理循环骨架，不依赖任何高级封装：

import torch from transformers import AutoTokenizer, AutoModelForCausalLM class SimpleInferenceEngine: def __init__(self, model_name): self.tokenizer = AutoTokenizer.from_pretrained(model_name) self.model = AutoModelForCausalLM.from_pretrained( model_name, torch_dtype=torch.float16, device_map="auto" ) self.model.eval() @torch.no_grad() def generate(self, prompt, max_new_tokens=256, temperature=0.7): inputs = self.tokenizer(prompt, return_tensors="pt").to(self.model.device) input_ids = inputs["input_ids"] # 手动维护KV Cache past_key_values = None generated_ids = input_ids for _ in range(max_new_tokens): outputs = self.model( input_ids=generated_ids if past_key_values is None else generated_ids[:, -1:], past_key_values=past_key_values, use_cache=True ) logits = outputs.logits[:, -1, :] past_key_values = outputs.past_key_values # 温度采样 probs = torch.softmax(logits / temperature, dim=-1) next_token = torch.multinomial(probs, num_samples=1) generated_ids = torch.cat([generated_ids, next_token], dim=-1) # 停止条件 if next_token.item() == self.tokenizer.eos_token_id: break return self.tokenizer.decode(generated_ids[0], skip_special_tokens=True)

这段代码的关键在于手动管理KV Cache。第一次推理传入完整prompt，后续每次只传入上一个生成的token，同时把past_key_values传回去。这样计算量从O(n²)降到O(n)，长文本生成速度提升非常明显。

2.2 KV Cache的内存账怎么算

KV Cache不是免费的。它的显存占用公式是：

KV Cache大小 = 2 * batch_size * num_layers * num_heads * head_dim * seq_len * dtype_size

以LLaMA-7B为例，32层，32个head，head_dim=128，float16精度，seq_len=2048：

2 * 1 * 32 * 32 * 128 * 2048 * 2 bytes ≈ 1.07 GB

这只是一个请求的缓存。如果你并发10个请求，直接吃掉10GB显存。所以生产环境必须做动态批处理和PagedAttention这类优化。但那是后面的事，第一步先把这个账算明白。

实操心得：调试推理循环时，先把max_new_tokens设成8，打印每一步的next_token和对应的概率分布。你会直观看到模型是怎么"选词"的，这对理解采样参数帮助极大。

2.3 采样策略的取舍：贪心、温度、Top-p

贪心解码（每次选概率最大的token）适合确定性任务，比如信息抽取。但生成文本会重复、死板。温度采样引入随机性，温度越高越随机。Top-p（核采样）只从累积概率达到p的最小token集合中采样，比Top-k更灵活。

我的经验配置：

这些参数不是拍脑袋定的，是大量A/B测试的结果。你可以从这些基准出发，根据自己业务微调。

3. 检索增强：从零搭建一个不胡说的RAG

3.1 RAG的本质是信息检索加条件生成

大模型胡说的根本原因是它不知道答案，但又被训练成必须回答。RAG的思路很朴素：先把相关资料找出来，塞进prompt里，让模型基于资料回答。但就这么个朴素思路，工程上有一堆细节决定成败。

一个完整的RAG流程包括：文档加载、分块、向量化、存储、检索、重排、生成。每个环节都有坑。

3.2 分块策略：固定长度是最差的选择

新手最容易犯的错是按固定字符数分块。比如每500字切一刀。结果一个完整的段落被切成两半，检索出来语义不完整。

我推荐递归分块：先按段落分，段落太长再按句子分，句子还长再按字符分。同时保留一定的重叠区域（overlap），避免边界信息丢失。

def recursive_chunk(text, max_chunk_size=512, overlap=50): paragraphs = text.split("\n\n") chunks = [] current_chunk = "" for para in paragraphs: if len(current_chunk) + len(para) <= max_chunk_size: current_chunk += para + "\n\n" else: if current_chunk: chunks.append(current_chunk.strip()) # 处理超长段落 if len(para) > max_chunk_size: sentences = para.split("。") temp = "" for sent in sentences: if len(temp) + len(sent) <= max_chunk_size: temp += sent + "。" else: chunks.append(temp.strip()) temp = sent + "。" current_chunk = temp else: current_chunk = para + "\n\n" if current_chunk: chunks.append(current_chunk.strip()) return chunks

重叠区域的作用是：如果答案刚好在切分边界上，至少有一个块包含完整信息。

3.3 向量化模型的选择：不是越大越好

向量化模型（embedding model）的选择直接决定检索质量。但参数大不等于效果好。bge-large在某些中文任务上不如bge-small，因为小模型在领域数据上更容易微调。

我的选型原则：

• 通用场景：先用bge-base-zh或text-embedding-3-small打底
• 垂直领域：拿几百条领域数据微调一个小模型，效果提升明显
• 多语言：paraphrase-multilingual-MiniLM-L12-v2性价比高

注意：向量维度不是越高越好。768维和1024维在检索准确率上差异很小，但存储和计算成本差一倍。先跑通流程，再考虑升级。

3.4 检索之后必须重排

向量检索是粗筛，Top-20里可能只有3-4条真正相关。直接塞给模型，噪声太大。重排模型（reranker）对粗筛结果做精排，把最相关的放到前面。

from sentence_transformers import CrossEncoder reranker = CrossEncoder("BAAI/bge-reranker-base") def retrieve_and_rerank(query, chunks, top_k=5): # 粗筛：向量检索拿Top-20 coarse_results = vector_search(query, chunks, top_k=20) # 精排：CrossEncoder打分 pairs = [[query, chunk] for chunk in coarse_results] scores = reranker.predict(pairs) # 按分数排序取Top-5 ranked = sorted(zip(coarse_results, scores), key=lambda x: x[1], reverse=True) return [chunk for chunk, _ in ranked[:top_k]]

重排的代价是延迟增加，但准确率提升通常值得。如果延迟敏感，可以用小模型做重排，或者只对Top-10重排。

3.5 生成阶段的prompt模板

检索到的内容怎么塞给模型，也有讲究。我的模板：

你是一个严谨的助手。请仅基于以下参考资料回答问题。 如果参考资料中没有相关信息，请直接说"根据现有资料无法回答"。 参考资料： {context} 问题：{question} 回答：

关键点是明确告诉模型可以拒绝回答。不加这句话，模型会强行编造。加了之后，拒答率上升，但幻觉率大幅下降。

4. Agent编排：让模型学会用工具

4.1 Agent的本质是循环加工具调用

Agent不是什么神秘东西。它的核心就是一个循环：观察-思考-行动-再观察。模型根据当前状态决定下一步做什么，调用工具，拿到结果，继续循环，直到任务完成。

一个最简Agent循环：

class SimpleAgent: def __init__(self, llm, tools): self.llm = llm self.tools = {tool.name: tool for tool in tools} self.max_steps = 10 def run(self, task): history = [{"role": "user", "content": task}] for step in range(self.max_steps): # 模型决定下一步 response = self.llm.chat(history, tools=self.tools) # 如果没有工具调用，直接返回 if not response.tool_calls: return response.content # 执行工具调用 for tool_call in response.tool_calls: tool_name = tool_call.function.name tool_args = json.loads(tool_call.function.arguments) try: result = self.tools[tool_name].execute(**tool_args) except Exception as e: result = f"工具执行失败：{str(e)}" history.append({ "role": "tool", "tool_call_id": tool_call.id, "content": str(result) }) return "达到最大步数限制，任务未完成"

这个循环看起来简单，但生产环境要处理的问题很多：工具调用失败怎么办、模型陷入死循环怎么办、并发工具调用怎么处理、上下文超长怎么截断。

4.2 工具设计的三条铁律

第一，工具描述要像给实习生写文档。模型不知道你的工具怎么用，全靠描述。描述里要写清楚：这个工具做什么、参数是什么类型、什么情况下用、什么情况下不用。

第二，工具粒度要适中。太细，模型要调很多次；太粗，模型不知道怎么组合。比如"查询数据库"太粗，"查询用户表按ID"又太细。合理的是"根据用户ID查询用户基本信息"。

第三，工具要幂等。模型可能重复调用同一个工具。如果工具不幂等，会产生副作用。查询类工具天然幂等，写入类工具要加去重逻辑。

4.3 上下文管理：Agent的隐形杀手

Agent循环跑几轮之后，上下文会爆炸。每轮的工具调用结果都塞进history，很快就超长。必须做上下文压缩。

我的策略是分层：

• 最近3轮：完整保留，包括工具调用细节
• 3-10轮：只保留工具调用的摘要，不保留原始结果
• 10轮以上：只保留任务目标和当前状态

def compress_history(history, max_recent=3): if len(history) <= max_recent * 2: return history recent = history[-(max_recent * 2):] older = history[:-(max_recent * 2)] # 对旧历史做摘要 summary = summarize_tool_calls(older) return [{"role": "system", "content": f"之前的操作摘要：{summary}"}] + recent

实操心得：Agent调试时，把每一步的response.tool_calls和工具返回结果打印出来。你会清晰看到模型是怎么"想"的。很多时候模型犯错是因为工具描述有歧义，而不是模型能力不行。

4.4 错误处理与重试机制

工具调用失败是常态。网络超时、参数格式错误、权限不足，各种问题。Agent必须能处理这些。

我的做法是：工具层做重试，Agent层做降级。工具内部对可重试错误（如超时）自动重试3次。如果最终失败，返回结构化错误信息给Agent。Agent看到错误后，可以选择换一个工具，或者向用户求助。

def execute_with_retry(tool, args, max_retries=3): for attempt in range(max_retries): try: return tool.execute(**args) except RetryableError as e: if attempt == max_retries - 1: return {"error": str(e), "retryable": True} time.sleep(2 ** attempt) except FatalError as e: return {"error": str(e), "retryable": False}

关键是区分可重试和不可重试错误。参数错误重试一万次也没用，直接返回让Agent调整。

5. 性能与成本：从能跑到跑得起的距离

5.1 延迟拆解：时间花在哪了

一个AI应用的端到端延迟包括：网络传输、预处理、推理、后处理。推理又分prefill和decode两个阶段。Prefill是处理输入prompt，计算密集；decode是逐token生成，内存密集。

优化延迟要对症下药：

流式输出是最容易见效的优化。用户不需要等完整结果，首token时间（TTFT）才是关键指标。把TTFT从2秒降到500ms，用户体验天差地别。

5.2 成本控制：Token就是钱

API调用按token计费，自己部署按GPU小时计费。不管哪种，成本都和token量正相关。控制成本的核心是减少无效token。

几个实用技巧：

• Prompt压缩：把冗长的系统提示精简，去掉重复说明
• 缓存：相同或相似请求缓存结果，命中率能到30%以上
• 分级模型：简单任务用小模型，复杂任务才用大模型
• 输出限制：设置合理的max_tokens，避免模型啰嗦

我做过一个统计：一个RAG应用，如果不做任何优化，每次请求平均消耗3000 token。做了prompt压缩和缓存之后，降到1200 token，成本直接砍掉60%。

5.3 量化：用精度换速度

模型量化是把float16权重转成int8或int4，减少显存占用和计算量。但量化会损失精度，需要评估。

我的经验：

• int8量化：精度损失很小（<1%），速度提升30-50%，推荐
• int4量化：精度损失明显（3-5%），速度提升60-80%，谨慎使用
• GPTQ/AWQ：比朴素量化效果好，适合生产

量化不是万能的。有些任务对精度敏感，比如数学推理，量化后错误率飙升。必须做A/B测试。

5.4 批处理：吞吐量的关键

单个请求跑得快没用，要看吞吐量。批处理把多个请求合并成一个batch，GPU利用率大幅提升。

但批处理有代价：排队延迟。请求要等凑够一批才能处理。动态批处理（continuous batching）解决了这个问题：不等整批完成，新请求随时加入。

class ContinuousBatcher: def __init__(self, max_batch_size=8, max_wait_ms=50): self.max_batch_size = max_batch_size self.max_wait_ms = max_wait_ms self.queue = [] async def add_request(self, request): self.queue.append(request) if len(self.queue) >= self.max_batch_size: return await self.process_batch() # 等待一小段时间看有没有更多请求 await asyncio.sleep(self.max_wait_ms / 1000) return await self.process_batch() async def process_batch(self): batch = self.queue[:self.max_batch_size] self.queue = self.queue[self.max_batch_size:] # 批量推理逻辑 return await self.batch_inference(batch)

max_wait_ms是个权衡：设大了吞吐量高但延迟高，设小了延迟低但吞吐量上不去。根据业务SLA调整。

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

6.1 模型输出乱码或重复

现象：生成的文本出现无意义重复，或者夹杂特殊字符。

排查思路：

1. 检查tokenizer和模型是否匹配。用错tokenizer会导致ID映射错误。
2. 检查是否触发了repetition_penalty。这个参数设太高会导致输出不连贯。
3. 检查eos_token_id是否正确。如果EOS token不对，模型不知道什么时候停。

解决：打印原始token ID和decode后的文本对比。如果ID正常但文本乱码，是tokenizer问题；如果ID就异常，是模型加载问题。

6.2 RAG检索不到相关内容

现象：明明知识库里有答案，但检索出来的都是不相关的。

排查思路：

1. 检查分块是否合理。块太大，语义稀释；块太小，信息不完整。
2. 检查embedding模型是否适合中文。有些英文模型在中文上表现很差。
3. 检查向量归一化。余弦相似度需要归一化向量。

解决：拿几个典型query，手动看Top-10检索结果。如果前10都不相关，是embedding问题；如果前10有相关的但排后面，是排序问题，加重排。

6.3 Agent陷入死循环

现象：Agent反复调用同一个工具，或者在不同工具之间来回跳。

排查思路：

1. 检查工具描述是否有歧义。模型可能误解了工具用途。
2. 检查是否有终止条件。没有明确的"任务完成"信号，模型会一直尝试。
3. 检查上下文是否包含矛盾信息。

解决：设置max_steps硬限制。同时在prompt里明确写"如果已经获得足够信息，请直接给出最终答案，不要再调用工具"。

6.4 显存溢出（OOM）

现象：推理过程中CUDA out of memory。

排查思路：

1. 计算KV Cache大小，看是否超过显存。
2. 检查batch size是否过大。
3. 检查是否有内存泄漏（past_key_values没释放）。

解决：减小batch size，启用梯度检查点（训练时），使用量化模型，或者升级硬件。软件层面，确保每次推理后释放不需要的tensor。

6.5 流式输出卡顿

现象：流式输出时，token一个一个蹦，但中间有明显停顿。

排查思路：

1. 检查是否在每个token后都做了同步操作。
2. 检查网络缓冲区设置。
3. 检查后处理逻辑是否阻塞。

解决：把后处理异步化，不要阻塞token流。使用yield而不是return。确保网络层没有不必要的缓冲。

避坑技巧：生产环境一定要加超时控制。不管是API调用还是自己推理，都要设超时。我见过太多因为一个请求卡死导致整个服务不可用的事故。

7. 从能跑到好用：我的工程化清单

把上面这些东西串起来，一个可用的AI工程系统需要这些组件：

• 推理层：支持动态批处理、KV Cache管理、流式输出
• 检索层：递归分块、向量索引、重排
• 编排层：Agent循环、工具注册、上下文压缩
• 监控层：延迟、吞吐量、错误率、token消耗
• 成本层：缓存、分级模型、量化

每个组件都不复杂，但组合起来需要仔细设计接口。我的建议是先跑通最小闭环，再逐个优化。不要一上来就追求完美架构，那只会让你永远停留在设计阶段。

最后分享一个我踩过的坑：早期做RAG时，我把所有文档一股脑塞进向量库，没有做元数据过滤。结果用户问"最新的政策是什么"，检索出来的全是三年前的旧文档。后来加了时间戳过滤和版本管理，问题才解决。元数据不是可选项，是必选项。