企业官网建设流程全解析

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 为什么我没用LangChain

LangChain刚出来的时候我试过，确实能快速搭出原型，但上生产的时候遇到了几个问题。一是抽象层次太多，出问题的时候排查链路很长，一个简单的调用要翻好几层源码才能定位。二是版本迭代太快，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，比pip+requirements.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 常见问题速查表

5.5 几个只有踩过才知道的坑

第一个坑：prompt里的变量没转义。用户输入如果包含特殊字符，可能破坏prompt结构。我遇到过一次用户输入里带了"""，直接把prompt的边界搞乱了。解决方案是对所有变量做转义，或者用结构化格式（比如JSON）传参。

第二个坑：异步上下文里的异常被吞掉。asyncio.gather默认会吞掉异常，导致问题被隐藏。要用return_exceptions=True然后手动检查。

第三个坑：缓存穿透。如果大量请求的key都不在缓存里，会全部打到模型。解决方案是对不存在的key也缓存一个空结果，设置较短的过期时间。

第四个坑：模型版本升级导致输出变化。模型提供商升级模型版本后，同样的prompt可能得到不同的输出。解决方案是锁定模型版本，升级前做回归测试。

6. 性能优化的几个实战技巧

6.1 缓存策略的分层设计

缓存是降低延迟和成本最有效的手段。我设计了三层缓存：精确缓存（完全相同的输入直接返回缓存结果）、语义缓存（语义相似的输入返回缓存结果）、片段缓存（prompt的固定部分缓存）。

精确缓存用Redis，key是输入的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能力建一个"体检"接口。这个接口用一组固定的测试用例跑一遍，输出各项指标。每次发版前跑一下，能快速发现回归问题。这个习惯帮我避免了好几次线上事故。