1. 从零搭建AI工程能力一个后端老兵的踩坑与重构实录ai-engineering-from-scratch这个标题第一次看到的时候我以为是又一个教你怎么调API的速成教程。点进去翻了翻发现它讲的是更底层的东西——怎么从零开始把AI能力真正工程化地落地到业务系统里。这跟我在过去两年里做的事情高度重合所以我想借这个标题聊聊我自己从零搭建AI工程体系的全过程包括技术选型的纠结、架构设计的取舍、以及那些只有真正踩过坑才知道的细节。先说清楚这个项目适合谁看。如果你是一个后端工程师或者技术负责人手头有一个业务场景想接入AI能力但不确定从哪开始、用什么框架、怎么保证稳定性那这篇内容就是写给你的。如果你只是想调个OpenAI的接口做个demo那可能不太需要往下看。这里讲的是工程化落地不是玩具项目。我自己的背景是做了八年后端主要写Java和Go两年前开始接触AI相关的工程化工作。第一个项目是把一个客服系统从纯规则引擎改造成规则大模型的混合架构。当时踩的坑包括但不限于模型响应延迟导致接口超时、并发上来之后token消耗失控、prompt版本管理混乱、以及最要命的——线上出了badcase根本不知道怎么排查。这些问题逼着我从零开始重新思考AI工程的架构也才有了后来这套相对完整的方案。2. 整体架构设计与技术选型思路2.1 为什么不能直接调API了事很多人刚开始做AI工程的时候第一反应就是写个函数调一下模型接口拿到结果返回给前端。这个做法在demo阶段没问题但一旦上生产就会暴露三个致命问题。第一个是延迟不可控。大模型的响应时间从几百毫秒到十几秒都有可能取决于输入长度、模型负载、网络状况。如果你的业务接口有超时限制比如网关层设了3秒那直接调用大概率会超时。我第一个项目就吃过这个亏客服系统前端等5秒没响应用户就开始狂点结果触发了一堆重复请求。第二个是成本不可控。每次调用都是真金白银如果没有缓存、没有限流、没有降级策略一个异常流量就能把你的预算烧光。我见过一个团队因为忘记加频率限制被爬虫刷了一晚上第二天账单出来直接傻眼。第三个是质量不可控。模型输出是概率性的同样的输入可能得到不同的输出。如果没有后处理、没有校验、没有fallback机制badcase会直接暴露给用户。所以ai-engineering-from-scratch的核心不是怎么调模型而是怎么在模型外面包一层工程化的壳让AI能力变得可观测、可控制、可降级。2.2 分层架构的取舍我最终采用的架构分成四层从下往上依次是模型接入层、能力编排层、业务适配层、监控治理层。模型接入层负责屏蔽不同模型提供商的差异。这一层的关键是定义一套统一的接口规范不管是OpenAI、Claude还是本地部署的开源模型都通过同一套抽象接口调用。这样做的好处是切换模型时业务代码不用改只需要改配置。我试过在项目中期从GPT-3.5切到GPT-4因为有了这层抽象切换只花了半天时间。能力编排层是核心负责把多个模型调用、工具调用、规则判断组合成一个完整的AI能力。比如一个智能客服回复的能力可能需要先做意图识别小模型再检索知识库向量搜索最后生成回复大模型。这一层我用的是自己写的一个轻量级编排引擎没有用LangChain原因后面会详细说。业务适配层负责把AI能力适配到具体的业务接口。这一层处理的是业务逻辑相关的事情比如参数校验、权限控制、结果格式化。监控治理层贯穿所有层负责日志、指标、追踪、告警。这一层是最容易被忽视但最重要的没有它你就是在裸奔。2.3 为什么我没用LangChainLangChain刚出来的时候我试过确实能快速搭出原型但上生产的时候遇到了几个问题。一是抽象层次太多出问题的时候排查链路很长一个简单的调用要翻好几层源码才能定位。二是版本迭代太快API经常变升级一次就要改一堆代码。三是性能开销对于延迟敏感的场景不太友好。我最后选择自己写编排引擎核心代码大概两千行左右包括DAG定义、节点执行、上下文管理、错误处理。自己写的好处是完全可控出问题能快速定位性能也能优化到极致。当然这需要你对AI工程的各个环节有比较深的理解如果是刚入门用LangChain快速起步也没问题但要做好后期重构的准备。3. 核心模块的详细实现与关键细节3.1 模型接入层的统一抽象模型接入层的核心是定义一个ModelProvider接口包含chat、embedding、rerank等基本方法。每个模型提供商实现这个接口业务代码只依赖接口不依赖具体实现。class ModelProvider(ABC): abstractmethod async def chat(self, messages: List[Message], **kwargs) - ChatResponse: pass abstractmethod async def embedding(self, texts: List[str]) - List[List[float]]: pass这里有几个关键细节。第一是超时控制每个请求必须设置超时时间而且要有重试机制。我的做法是设置两级超时单次请求超时和总超时。单次超时到了就重试总超时到了就直接返回降级结果。第二是重试策略不是所有错误都值得重试。网络错误、限流错误可以重试参数错误、内容审核错误重试也没用。我实现了一个错误分类器根据错误类型决定是否重试以及重试间隔。第三是token计数每次调用都要记录消耗的token数用于成本核算和限流。不同模型的token计算方式不一样需要在各自的实现里处理。第四是流式响应对于需要流式输出的场景比如打字机效果要支持SSE或者WebSocket。流式响应的错误处理比普通请求复杂因为可能在流的中途出错需要设计好中断和恢复机制。3.2 能力编排层的DAG设计能力编排层我用的是DAG有向无环图模型每个节点是一个执行单元边表示依赖关系。一个典型的客服回复能力对应的DAG是这样的输入预处理 - 意图识别 - 并行执行知识检索 用户画像获取 - 回复生成 - 后处理 - 输出。节点类型我定义了四种模型节点调用大模型、工具节点调用外部API或函数、逻辑节点条件判断、循环、转换节点数据格式转换。每种节点有统一的执行接口接收上下文返回结果。class Node(ABC): abstractmethod async def execute(self, context: Context) - NodeResult: pass上下文Context是一个贯穿整个DAG执行过程的数据结构存储了输入参数、中间结果、执行状态等。我设计Context的时候参考了请求作用域的思路每个请求一个独立的Context避免并发时的数据污染。DAG的执行引擎我实现了一个简单的调度器支持串行、并行、条件分支。并行执行是关键因为很多步骤之间没有依赖关系并行能显著降低总延迟。比如知识检索和用户画像获取就可以并行实测下来能省30%左右的时间。3.3 业务适配层的参数处理业务适配层看起来简单其实有很多细节。首先是输入校验用户输入可能包含敏感信息、超长文本、特殊字符需要在进入AI流程之前处理掉。我实现了一个输入清洗管道包括长度截断、敏感词过滤、格式规范化。其次是结果后处理模型输出不能直接返回给用户需要做几件事格式校验比如要求JSON输出但模型返回了markdown、内容审核过滤不当内容、业务规则校验比如金额不能为负。后处理失败时要走降级逻辑返回预设的兜底回复。第三是多轮对话管理如果业务场景需要多轮交互要维护对话历史。对话历史不能无限增长需要做截断或者摘要。我的做法是保留最近N轮完整对话更早的对话用模型生成摘要。3.4 监控治理层的可观测性建设监控治理层是我花时间最多的地方因为吃过亏。核心要监控的指标包括延迟P50、P95、P99、成功率、token消耗、缓存命中率、降级触发次数。日志方面每次AI调用都要记录完整的请求和响应包括prompt、模型输出、耗时、token数。这些日志量很大需要做采样和异步写入。我用的是结构化日志方便后续做分析和回放。追踪方面我用OpenTelemetry做全链路追踪每个DAG节点是一个span能看到每个节点的耗时和状态。出问题的时候能快速定位是哪个环节慢了或者错了。告警方面设置了几个关键阈值P99延迟超过5秒告警、成功率低于95%告警、token消耗超过预算告警。告警要能区分是偶发还是持续避免告警风暴。4. 实操过程中的关键步骤与配置4.1 环境准备与依赖管理项目初始化的时候我建议用Python 3.10以上版本因为用到了不少异步特性。依赖管理用poetry比piprequirements.txt好用很多尤其是处理依赖冲突的时候。poetry init poetry add fastapi uvicorn httpx pydantic opentelemetry-api poetry add openai anthropic --optional关键依赖包括FastAPI做Web框架、httpx做异步HTTP客户端、pydantic做数据校验、opentelemetry做追踪。模型SDK按需安装不要全部装不然依赖冲突很麻烦。配置管理用环境变量加配置文件的方式。敏感信息API key只放环境变量其他配置放YAML文件。配置要支持热更新改配置不用重启服务。4.2 模型接入层的实现步骤第一步定义统一的接口和数据结构。Message、ChatResponse、EmbeddingResponse这些数据结构要设计好考虑扩展性。第二步实现各个提供商的适配器。以OpenAI为例需要处理API key认证、请求构造、响应解析、错误映射。错误映射很关键要把不同提供商的错误码映射到统一的错误类型。第三步实现重试和超时逻辑。我用的是tenacity库做重试配置了指数退避策略。超时用httpx的timeout参数控制。第四步实现token计数。OpenAI的响应里会返回token数直接用就行。有些模型不返回需要用tiktoken之类的库自己算。第五步实现流式响应。用httpx的stream方法逐块读取SSE数据解析后yield出去。4.3 编排引擎的实现步骤第一步定义DAG的数据结构。我用的是邻接表表示每个节点记录自己的依赖节点。第二步实现拓扑排序确定节点的执行顺序。有环的图要检测出来并报错。第三步实现执行调度器。支持串行和并行并行用asyncio.gather。第四步实现上下文管理。Context要支持嵌套子DAG能访问父DAG的上下文。第五步实现错误处理。节点执行失败时根据配置决定是中断整个DAG还是跳过继续。4.4 监控埋点的实现步骤第一步集成OpenTelemetry SDK配置exporter。第二步在每个DAG节点执行前后打span记录关键属性。第三步实现指标收集用prometheus_client暴露指标。第四步配置告警规则接入告警系统。第五步搭建日志收集和分析管道我用的是ELK。5. 常见问题与排查技巧实录5.1 延迟突然飙升怎么排查延迟飙升是最常见的问题排查思路是从外到内逐层定位。先看是整体延迟高还是个别请求高如果是个别请求看这些请求有什么共同特征比如输入特别长、调用了某个特定模型。如果是整体延迟高看是模型侧慢还是工程侧慢。模型侧慢的话看模型提供商的status page或者自己做个简单的探测请求。工程侧慢的话看DAG各节点的耗时定位到具体节点。常见原因包括连接池不够、DNS解析慢、序列化开销大、GC停顿。我遇到过一次延迟飙升最后定位到是日志写入阻塞了主流程。因为日志是同步写的磁盘IO满了之后整个请求就卡住了。改成异步写入后问题解决。5.2 token消耗失控怎么控制token消耗失控通常有三个原因请求量突增、单次请求token过多、缓存失效。请求量突增的话加限流。我用的是令牌桶算法按用户和按接口两个维度限流。单次请求token过多的话做输入截断和输出限制。输入超过阈值就截断或者摘要输出设置max_tokens。缓存失效的话检查缓存key的设计。缓存key要包含所有影响输出的因素比如prompt版本、模型版本、输入内容。我见过因为缓存key没包含模型版本切换模型后还在用旧缓存的bug。5.3 模型输出不稳定怎么处理模型输出不稳定是概率性模型的固有特性工程上能做的是降低方差和兜底。降低方差的手段包括降低temperature、用few-shot示例固定输出格式、用JSON mode强制结构化输出、多次采样取多数。兜底的手段包括输出格式校验失败时重试、重试失败返回预设回复、关键业务场景用规则引擎兜底。我一般会设置一个重试次数上限比如3次3次都失败就降级。降级策略要根据业务场景设计客服场景可以返回抱歉我暂时无法回答交易场景必须走人工审核。5.4 常见问题速查表问题现象可能原因排查方法解决方案接口超时模型响应慢看模型调用耗时加超时降级成本超预算请求量或token失控看token消耗趋势限流缓存输出格式错模型不稳定看badcase日志加校验重试并发上不去连接池或锁竞争看线程/协程状态调连接池异步化内存泄漏上下文未释放看内存增长曲线检查Context生命周期5.5 几个只有踩过才知道的坑第一个坑prompt里的变量没转义。用户输入如果包含特殊字符可能破坏prompt结构。我遇到过一次用户输入里带了直接把prompt的边界搞乱了。解决方案是对所有变量做转义或者用结构化格式比如JSON传参。第二个坑异步上下文里的异常被吞掉。asyncio.gather默认会吞掉异常导致问题被隐藏。要用return_exceptionsTrue然后手动检查。第三个坑缓存穿透。如果大量请求的key都不在缓存里会全部打到模型。解决方案是对不存在的key也缓存一个空结果设置较短的过期时间。第四个坑模型版本升级导致输出变化。模型提供商升级模型版本后同样的prompt可能得到不同的输出。解决方案是锁定模型版本升级前做回归测试。6. 性能优化的几个实战技巧6.1 缓存策略的分层设计缓存是降低延迟和成本最有效的手段。我设计了三层缓存精确缓存完全相同的输入直接返回缓存结果、语义缓存语义相似的输入返回缓存结果、片段缓存prompt的固定部分缓存。精确缓存用Rediskey是输入的hash。语义缓存用向量数据库把输入embedding后做相似度搜索。片段缓存是把prompt拆成固定部分和可变部分固定部分的KV cache可以复用。语义缓存的阈值要调太高了命中率低太低了可能返回不相关的结果。我一般从0.95开始调根据业务反馈调整。6.2 批处理与并发控制对于批量任务比如批量生成摘要用批处理能显著提升吞吐。OpenAI的batch API比实时API便宜一半适合对延迟不敏感的场景。并发控制用信号量限制同时进行的模型调用数。并发数不是越高越好太高了会触发限流反而降低吞吐。我一般从10开始压测找到吞吐的拐点。6.3 模型选择的性价比权衡不是所有场景都需要用最贵的模型。我的做法是把任务分级简单任务用小模型比如意图识别用fine-tune的小模型复杂任务用大模型。实测下来能省60%以上的成本。还有一个技巧是用模型路由根据输入的特征动态选择模型。比如输入短且简单就用小模型输入长且复杂就用大模型。路由规则可以基于规则也可以用一个小的分类模型。7. 从零搭建的完整流程回顾如果你现在要从零开始搭建AI工程能力我建议按这个顺序来。第一周搭基础框架。把模型接入层和最简单的编排能力做出来能跑通一个端到端的流程。不要追求完美先跑起来。第二周加监控和日志。这是最重要的基础设施越早加越好。没有监控你就是在盲人摸象。第三周加缓存和限流。这两个是控制成本和延迟的关键。第四周加降级和容错。生产环境什么都会发生要有兜底方案。之后就是持续迭代根据业务反馈优化prompt、调整模型、优化性能。我在实际项目中发现最难的不是技术实现而是建立一套可持续迭代的机制。AI工程和传统工程最大的区别是模型的行为是不确定的你需要一套机制来持续观测、评估、优化。这套机制包括badcase收集流程、prompt版本管理、A/B测试框架、回归测试集。这些东西建起来之后迭代效率会高很多。最后分享一个小技巧给每个AI能力建一个体检接口。这个接口用一组固定的测试用例跑一遍输出各项指标。每次发版前跑一下能快速发现回归问题。这个习惯帮我避免了好几次线上事故。
