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_dtypetorch.float16, device_mapauto ) self.model.eval() torch.no_grad() def generate(self, prompt, max_new_tokens256, temperature0.7): inputs self.tokenizer(prompt, return_tensorspt).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_idsgenerated_ids if past_key_values is None else generated_ids[:, -1:], past_key_valuespast_key_values, use_cacheTrue ) logits outputs.logits[:, -1, :] past_key_values outputs.past_key_values # 温度采样 probs torch.softmax(logits / temperature, dim-1) next_token torch.multinomial(probs, num_samples1) 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_tokensTrue)这段代码的关键在于手动管理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个headhead_dim128float16精度seq_len20482 * 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更灵活。我的经验配置任务类型temperaturetop_p说明代码生成0.20.95低温度保证语法正确创意写作0.80.9高温度增加多样性信息抽取0.01.0贪心要确定性对话系统0.70.9平衡多样性和连贯性这些参数不是拍脑袋定的是大量A/B测试的结果。你可以从这些基准出发根据自己业务微调。3. 检索增强从零搭建一个不胡说的RAG3.1 RAG的本质是信息检索加条件生成大模型胡说的根本原因是它不知道答案但又被训练成必须回答。RAG的思路很朴素先把相关资料找出来塞进prompt里让模型基于资料回答。但就这么个朴素思路工程上有一堆细节决定成败。一个完整的RAG流程包括文档加载、分块、向量化、存储、检索、重排、生成。每个环节都有坑。3.2 分块策略固定长度是最差的选择新手最容易犯的错是按固定字符数分块。比如每500字切一刀。结果一个完整的段落被切成两半检索出来语义不完整。我推荐递归分块先按段落分段落太长再按句子分句子还长再按字符分。同时保留一定的重叠区域overlap避免边界信息丢失。def recursive_chunk(text, max_chunk_size512, overlap50): 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_k5): # 粗筛向量检索拿Top-20 coarse_results vector_search(query, chunks, top_k20) # 精排CrossEncoder打分 pairs [[query, chunk] for chunk in coarse_results] scores reranker.predict(pairs) # 按分数排序取Top-5 ranked sorted(zip(coarse_results, scores), keylambda x: x[1], reverseTrue) 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, toolsself.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_recent3): 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_retries3): 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生成内存密集。优化延迟要对症下药阶段优化手段预期收益网络传输边缘部署、CDN减少50-200ms预处理异步化、缓存减少10-50msPrefill批处理、量化减少30-60%DecodeKV Cache优化、投机采样减少40-70%后处理流式输出感知延迟降低流式输出是最容易见效的优化。用户不需要等完整结果首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 批处理吞吐量的关键单个请求跑得快没用要看吞吐量。批处理把多个请求合并成一个batchGPU利用率大幅提升。但批处理有代价排队延迟。请求要等凑够一批才能处理。动态批处理continuous batching解决了这个问题不等整批完成新请求随时加入。class ContinuousBatcher: def __init__(self, max_batch_size8, max_wait_ms50): 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 模型输出乱码或重复现象生成的文本出现无意义重复或者夹杂特殊字符。排查思路检查tokenizer和模型是否匹配。用错tokenizer会导致ID映射错误。检查是否触发了repetition_penalty。这个参数设太高会导致输出不连贯。检查eos_token_id是否正确。如果EOS token不对模型不知道什么时候停。解决打印原始token ID和decode后的文本对比。如果ID正常但文本乱码是tokenizer问题如果ID就异常是模型加载问题。6.2 RAG检索不到相关内容现象明明知识库里有答案但检索出来的都是不相关的。排查思路检查分块是否合理。块太大语义稀释块太小信息不完整。检查embedding模型是否适合中文。有些英文模型在中文上表现很差。检查向量归一化。余弦相似度需要归一化向量。解决拿几个典型query手动看Top-10检索结果。如果前10都不相关是embedding问题如果前10有相关的但排后面是排序问题加重排。6.3 Agent陷入死循环现象Agent反复调用同一个工具或者在不同工具之间来回跳。排查思路检查工具描述是否有歧义。模型可能误解了工具用途。检查是否有终止条件。没有明确的任务完成信号模型会一直尝试。检查上下文是否包含矛盾信息。解决设置max_steps硬限制。同时在prompt里明确写如果已经获得足够信息请直接给出最终答案不要再调用工具。6.4 显存溢出OOM现象推理过程中CUDA out of memory。排查思路计算KV Cache大小看是否超过显存。检查batch size是否过大。检查是否有内存泄漏past_key_values没释放。解决减小batch size启用梯度检查点训练时使用量化模型或者升级硬件。软件层面确保每次推理后释放不需要的tensor。6.5 流式输出卡顿现象流式输出时token一个一个蹦但中间有明显停顿。排查思路检查是否在每个token后都做了同步操作。检查网络缓冲区设置。检查后处理逻辑是否阻塞。解决把后处理异步化不要阻塞token流。使用yield而不是return。确保网络层没有不必要的缓冲。避坑技巧生产环境一定要加超时控制。不管是API调用还是自己推理都要设超时。我见过太多因为一个请求卡死导致整个服务不可用的事故。7. 从能跑到好用我的工程化清单把上面这些东西串起来一个可用的AI工程系统需要这些组件推理层支持动态批处理、KV Cache管理、流式输出检索层递归分块、向量索引、重排编排层Agent循环、工具注册、上下文压缩监控层延迟、吞吐量、错误率、token消耗成本层缓存、分级模型、量化每个组件都不复杂但组合起来需要仔细设计接口。我的建议是先跑通最小闭环再逐个优化。不要一上来就追求完美架构那只会让你永远停留在设计阶段。最后分享一个我踩过的坑早期做RAG时我把所有文档一股脑塞进向量库没有做元数据过滤。结果用户问最新的政策是什么检索出来的全是三年前的旧文档。后来加了时间戳过滤和版本管理问题才解决。元数据不是可选项是必选项。
从零手搓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_dtypetorch.float16, device_mapauto ) self.model.eval() torch.no_grad() def generate(self, prompt, max_new_tokens256, temperature0.7): inputs self.tokenizer(prompt, return_tensorspt).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_idsgenerated_ids if past_key_values is None else generated_ids[:, -1:], past_key_valuespast_key_values, use_cacheTrue ) logits outputs.logits[:, -1, :] past_key_values outputs.past_key_values # 温度采样 probs torch.softmax(logits / temperature, dim-1) next_token torch.multinomial(probs, num_samples1) 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_tokensTrue)这段代码的关键在于手动管理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个headhead_dim128float16精度seq_len20482 * 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更灵活。我的经验配置任务类型temperaturetop_p说明代码生成0.20.95低温度保证语法正确创意写作0.80.9高温度增加多样性信息抽取0.01.0贪心要确定性对话系统0.70.9平衡多样性和连贯性这些参数不是拍脑袋定的是大量A/B测试的结果。你可以从这些基准出发根据自己业务微调。3. 检索增强从零搭建一个不胡说的RAG3.1 RAG的本质是信息检索加条件生成大模型胡说的根本原因是它不知道答案但又被训练成必须回答。RAG的思路很朴素先把相关资料找出来塞进prompt里让模型基于资料回答。但就这么个朴素思路工程上有一堆细节决定成败。一个完整的RAG流程包括文档加载、分块、向量化、存储、检索、重排、生成。每个环节都有坑。3.2 分块策略固定长度是最差的选择新手最容易犯的错是按固定字符数分块。比如每500字切一刀。结果一个完整的段落被切成两半检索出来语义不完整。我推荐递归分块先按段落分段落太长再按句子分句子还长再按字符分。同时保留一定的重叠区域overlap避免边界信息丢失。def recursive_chunk(text, max_chunk_size512, overlap50): 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_k5): # 粗筛向量检索拿Top-20 coarse_results vector_search(query, chunks, top_k20) # 精排CrossEncoder打分 pairs [[query, chunk] for chunk in coarse_results] scores reranker.predict(pairs) # 按分数排序取Top-5 ranked sorted(zip(coarse_results, scores), keylambda x: x[1], reverseTrue) 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, toolsself.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_recent3): 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_retries3): 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生成内存密集。优化延迟要对症下药阶段优化手段预期收益网络传输边缘部署、CDN减少50-200ms预处理异步化、缓存减少10-50msPrefill批处理、量化减少30-60%DecodeKV Cache优化、投机采样减少40-70%后处理流式输出感知延迟降低流式输出是最容易见效的优化。用户不需要等完整结果首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 批处理吞吐量的关键单个请求跑得快没用要看吞吐量。批处理把多个请求合并成一个batchGPU利用率大幅提升。但批处理有代价排队延迟。请求要等凑够一批才能处理。动态批处理continuous batching解决了这个问题不等整批完成新请求随时加入。class ContinuousBatcher: def __init__(self, max_batch_size8, max_wait_ms50): 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 模型输出乱码或重复现象生成的文本出现无意义重复或者夹杂特殊字符。排查思路检查tokenizer和模型是否匹配。用错tokenizer会导致ID映射错误。检查是否触发了repetition_penalty。这个参数设太高会导致输出不连贯。检查eos_token_id是否正确。如果EOS token不对模型不知道什么时候停。解决打印原始token ID和decode后的文本对比。如果ID正常但文本乱码是tokenizer问题如果ID就异常是模型加载问题。6.2 RAG检索不到相关内容现象明明知识库里有答案但检索出来的都是不相关的。排查思路检查分块是否合理。块太大语义稀释块太小信息不完整。检查embedding模型是否适合中文。有些英文模型在中文上表现很差。检查向量归一化。余弦相似度需要归一化向量。解决拿几个典型query手动看Top-10检索结果。如果前10都不相关是embedding问题如果前10有相关的但排后面是排序问题加重排。6.3 Agent陷入死循环现象Agent反复调用同一个工具或者在不同工具之间来回跳。排查思路检查工具描述是否有歧义。模型可能误解了工具用途。检查是否有终止条件。没有明确的任务完成信号模型会一直尝试。检查上下文是否包含矛盾信息。解决设置max_steps硬限制。同时在prompt里明确写如果已经获得足够信息请直接给出最终答案不要再调用工具。6.4 显存溢出OOM现象推理过程中CUDA out of memory。排查思路计算KV Cache大小看是否超过显存。检查batch size是否过大。检查是否有内存泄漏past_key_values没释放。解决减小batch size启用梯度检查点训练时使用量化模型或者升级硬件。软件层面确保每次推理后释放不需要的tensor。6.5 流式输出卡顿现象流式输出时token一个一个蹦但中间有明显停顿。排查思路检查是否在每个token后都做了同步操作。检查网络缓冲区设置。检查后处理逻辑是否阻塞。解决把后处理异步化不要阻塞token流。使用yield而不是return。确保网络层没有不必要的缓冲。避坑技巧生产环境一定要加超时控制。不管是API调用还是自己推理都要设超时。我见过太多因为一个请求卡死导致整个服务不可用的事故。7. 从能跑到好用我的工程化清单把上面这些东西串起来一个可用的AI工程系统需要这些组件推理层支持动态批处理、KV Cache管理、流式输出检索层递归分块、向量索引、重排编排层Agent循环、工具注册、上下文压缩监控层延迟、吞吐量、错误率、token消耗成本层缓存、分级模型、量化每个组件都不复杂但组合起来需要仔细设计接口。我的建议是先跑通最小闭环再逐个优化。不要一上来就追求完美架构那只会让你永远停留在设计阶段。最后分享一个我踩过的坑早期做RAG时我把所有文档一股脑塞进向量库没有做元数据过滤。结果用户问最新的政策是什么检索出来的全是三年前的旧文档。后来加了时间戳过滤和版本管理问题才解决。元数据不是可选项是必选项。
