1. 从零搭建AI工程能力为什么我劝你别一上来就调包这两年AI应用开发的门槛肉眼可见地降低了随便拉个框架、调个API就能跑出一个能对话的Demo。但我带过的新人里十个有八个卡在同一个地方Demo跑通了一上真实业务就崩。要么是响应慢得离谱要么是并发一上来就报错要么是模型输出不稳定导致下游逻辑全乱。问题的根子不在模型本身而在于大多数人跳过了“AI工程”这个环节直接从“调包”跳到了“上线”。“ai-engineering-from-scratch”这个标题我理解它想说的不是“从零训练一个大模型”——那是研究机构干的事普通人既没算力也没数据。它真正指向的是一套从零构建AI应用工程体系的路径怎么把一个大模型的能力稳定、可控、可维护地接入到实际产品里。这中间涉及提示词管理、上下文组织、输出解析、错误重试、成本控制、效果评估等一系列工程问题。这些东西没有哪本教材会系统讲但每一个都是真实项目里绕不过去的坎。这篇文章适合三类人看一是刚转行做AI应用开发、还在“调包”阶段徘徊的工程师二是有后端或前端经验、想把自己的工程能力迁移到AI场景的开发者三是带团队做AI产品、需要一套可落地工程规范的技术负责人。我会把整个从零搭建的过程拆成可复现的步骤每个环节都讲清楚“为什么这么做”以及“不这么做会怎样”。你不需要有机器学习背景但最好写过至少一个完整的项目知道什么是环境变量、什么是接口超时、什么是日志。2. 整体设计思路把AI能力当成一个“不靠谱的远程服务”来对待2.1 核心认知转变模型不是函数是概率服务很多从传统后端转过来的人最大的思维惯性是把模型调用当成一个普通函数输入确定输出就确定。但大模型本质是一个概率系统同样的输入两次调用可能给出不同的输出。这不是Bug是特性。你如果按“确定性函数”的思路去设计系统后面一定会被各种边界情况搞崩溃。我在实际项目里总结出一个原则把模型调用当成一个“不靠谱的远程服务”来对待。什么叫不靠谱它可能超时、可能返回格式不对、可能答非所问、可能突然涨价、可能某天接口挂掉。你的工程体系要做的就是在这个不靠谱的服务之上包一层可靠的壳。这层壳包括输入预处理、输出校验、失败重试、降级策略、成本监控。这跟当年我们对待第三方支付接口的思路是一样的只不过模型的不确定性更高、更隐蔽。2.2 分层架构为什么我不建议把逻辑全写在业务代码里我见过太多项目模型调用代码直接散落在各个业务函数里。今天在用户注册流程里加一句“帮我生成欢迎语”明天在订单模块里加一句“帮我总结评价”。三个月后想换个模型、想调一下提示词、想加个缓存发现要改几十个文件。这就是典型的“没有分层”的代价。从零搭建AI工程能力第一件事就是把AI相关的逻辑从业务逻辑里抽出来形成独立的一层。我通常分成四层接入层负责和模型服务通信处理鉴权、超时、重试、限流。这一层不关心业务只关心“怎么把请求发出去、把响应收回来”。编排层负责组织上下文、拼接提示词、管理多轮对话状态。这一层是AI应用的核心也是最容易出问题的地方。解析层负责把模型的自然语言输出转换成程序能用的结构化数据。这一层决定了你的系统能不能稳定运转。业务层真正调用AI能力的业务代码。这一层应该尽量薄只负责“什么时候调、拿到结果后干什么”。这么分的好处是当你想换模型时只改接入层当你想调提示词时只改编排层当你想改输出格式时只改解析层。业务层几乎不动。这就是工程化的价值。2.3 技术选型为什么我最终选了这套组合在从零搭建的过程中技术选型是最容易纠结的环节。我试过不少方案最后沉淀下来的组合是这样的环节选型理由语言PythonAI生态最全调试方便团队上手快模型接入官方SDK 自建封装官方SDK稳定但需要自己包一层统一接口提示词管理独立配置文件 版本号提示词是核心资产必须可追溯、可回滚输出解析Pydantic 正则兜底结构化校验强异常情况有兜底缓存Redis相同输入直接命中省成本省时间监控自建日志 成本统计必须知道每次调用花了多少钱、耗时多少这里重点说两个选型背后的逻辑。第一为什么提示词要独立管理。很多人把提示词硬编码在代码里改一个词就要重新部署。但提示词是需要反复迭代的今天加一句约束明天删一句示例如果每次都要走发布流程迭代效率极低。我通常把提示词放在独立的配置文件或配置中心里带上版本号每次调用记录用了哪个版本。这样既能快速迭代又能回溯“哪个版本效果最好”。第二为什么输出解析要用Pydantic。大模型的输出是自然语言但你的下游代码需要的是结构化数据。比如你让模型“提取用户评价中的情感倾向和关键词”它可能返回一段话也可能返回JSON还可能返回Markdown表格。如果你直接用字符串处理迟早会被各种格式搞疯。Pydantic的好处是你可以定义一个严格的数据模型让模型按这个模型输出然后用Pydantic做校验。校验不通过就触发重试或降级。这是保证系统稳定的关键一环。3. 核心细节解析从零搭建的五个关键环节3.1 环境准备与依赖管理别小看这一步从零开始第一步永远是环境。我见过太多项目因为环境问题卡住半天。Python项目我建议用venv或conda创建独立环境不要用系统Python。依赖管理用requirements.txt或pyproject.toml把版本号锁死。AI相关的库更新极快今天能跑的代码下周可能就因为某个库升级而报错。锁版本是保命手段。python -m venv venv source venv/bin/activate # Windows用 venv\Scripts\activate pip install openai pydantic redis python-dotenv pip freeze requirements.txt环境变量用.env文件管理不要把API Key硬编码在代码里。这不是安全问题是习惯问题。一旦你养成把密钥写死在代码里的习惯后面协作时一定会出事。# .env MODEL_API_KEYyour_key_here MODEL_BASE_URLhttps://api.example.com/v1 REDIS_URLredis://localhost:6379/0注意.env文件一定要加到.gitignore里。我见过有人把带密钥的.env提交到公开仓库结果被扫到后产生巨额账单。这种事一次就够你记住一辈子。3.2 模型接入层的封装统一接口隔离变化接入层的核心目标是让上层代码不感知具体用的是哪家模型。今天用A模型明天想换B模型上层代码一行不改。怎么做定义一个统一的接口。from abc import ABC, abstractmethod from pydantic import BaseModel class ModelResponse(BaseModel): content: str model: str tokens_used: int latency_ms: float class BaseModelClient(ABC): abstractmethod def chat(self, messages: list, **kwargs) - ModelResponse: pass然后针对不同模型写具体的实现类。每个实现类里处理该模型特有的鉴权、参数格式、错误码。上层只调用chat方法拿到统一的ModelResponse。这样换模型时只需要新增一个实现类改一下配置上层完全无感。重试逻辑也放在这一层。我的经验是超时重试最多两次间隔用指数退避。第一次等1秒第二次等2秒第三次等4秒。超过三次还没成功直接抛异常让上层处理。不要无限重试那只会让系统雪崩。import time from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) def call_with_retry(client, messages): return client.chat(messages)3.3 提示词工程不是写作文是写规格说明书很多人把提示词当成“跟模型聊天”这是最大的误区。在生产环境里提示词是一份规格说明书你要用最精确的语言告诉模型你的角色是什么、输入是什么、输出格式是什么、边界情况怎么处理。我通常把提示词拆成四个部分角色定义一句话说清楚模型扮演什么角色。比如“你是一个电商评论分析助手”。任务描述具体要做什么。比如“从用户评论中提取情感倾向和关键词”。输出格式严格定义输出结构。最好给一个JSON示例。约束条件什么不能做。比如“不要编造评论中没有的信息”。PROMPT_TEMPLATE 你是一个电商评论分析助手。 任务从用户评论中提取情感倾向和关键词。 输出格式严格按此JSON输出不要添加任何其他内容 {{ sentiment: positive | negative | neutral, keywords: [关键词1, 关键词2], confidence: 0.0 到 1.0 之间的浮点数 }} 约束 - 只提取评论中明确出现的信息不要推断。 - 如果评论无法判断情感sentiment填neutral。 - keywords最多5个按重要性排序。 用户评论{review} 这里有个细节JSON示例里的花括号要转义。Python的format方法会把{}当成占位符所以要用{{和}}。这个坑我踩过调试了半天才发现是转义问题。提示词版本管理也很重要。我通常在配置文件里给每个提示词一个版本号调用时记录用了哪个版本。这样当效果变差时可以快速定位是不是提示词改动导致的。3.4 输出解析与校验把自然语言变成程序能用的数据模型输出解析是AI工程里最容易被低估的环节。很多人觉得“模型返回JSON就行了”但实际生产中模型可能返回带Markdown代码块的JSON、可能返回多余的解释文字、可能字段名拼错、可能类型不对。你必须有一套健壮的解析流程。我的做法是三层解析第一层直接解析。尝试用json.loads解析原始输出。第二层提取解析。如果直接解析失败用正则提取{...}之间的内容再解析。第三层模型修复。如果还失败把原始输出和错误信息发给模型让它修复成正确格式。import json import re from pydantic import BaseModel, ValidationError class ReviewAnalysis(BaseModel): sentiment: str keywords: list[str] confidence: float def parse_output(raw: str) - ReviewAnalysis: # 第一层直接解析 try: data json.loads(raw) return ReviewAnalysis(**data) except (json.JSONDecodeError, ValidationError): pass # 第二层正则提取 match re.search(r\{.*\}, raw, re.DOTALL) if match: try: data json.loads(match.group()) return ReviewAnalysis(**data) except (json.JSONDecodeError, ValidationError): pass # 第三层抛异常由上层决定是否触发模型修复 raise ValueError(f无法解析模型输出: {raw[:200]})实操心得第三层“模型修复”要慎用。它虽然能提高成功率但会增加一次调用成本。我的经验是如果前两层解析失败率超过5%说明提示词写得有问题应该去优化提示词而不是依赖修复。3.5 缓存与成本控制省下来的都是利润AI应用的成本主要来自模型调用。每次调用都花钱而且价格不菲。缓存是最直接的省钱手段。相同的输入直接返回缓存结果不调模型。但缓存有个坑不是所有场景都适合缓存。比如对话场景同样的用户输入在不同上下文下含义不同不能简单缓存。我通常只对“无状态”的调用做缓存比如文本分类、信息提取、翻译。这些场景输入确定输出就确定缓存命中率高。import hashlib import redis import json redis_client redis.from_url(os.getenv(REDIS_URL)) def get_cache_key(prompt: str, model: str) - str: content f{model}:{prompt} return fai_cache:{hashlib.md5(content.encode()).hexdigest()} def cached_call(prompt: str, model: str, ttl: int 3600): key get_cache_key(prompt, model) cached redis_client.get(key) if cached: return json.loads(cached) result call_model(prompt, model) redis_client.setex(key, ttl, json.dumps(result)) return result除了缓存还要做成本监控。每次调用记录模型名、输入token数、输出token数、耗时、费用。这些数据积累起来你才能知道钱花在哪了、哪个功能最费钱、有没有优化空间。我通常用一个简单的日志表来记录每天跑个脚本统计。4. 实操过程从零到一搭建一个评论分析服务4.1 项目结构设计说了这么多理论不如直接上手搭一个。我们做一个电商评论分析服务输入一条评论输出情感倾向和关键词。这个场景足够简单但涵盖了AI工程的核心环节。项目结构这样设计ai-review-analyzer/ ├── .env ├── requirements.txt ├── config/ │ └── prompts.yaml ├── src/ │ ├── __init__.py │ ├── client/ │ │ ├── __init__.py │ │ ├── base.py │ │ └── openai_client.py │ ├── orchestrator/ │ │ ├── __init__.py │ │ └── review_analyzer.py │ ├── parser/ │ │ ├── __init__.py │ │ └── review_parser.py │ └── service/ │ ├── __init__.py │ └── analyze_service.py └── tests/ └── test_analyze.py这个结构对应前面说的四层client是接入层orchestrator是编排层parser是解析层service是业务层。config放提示词配置。4.2 提示词配置化config/prompts.yamlreview_analysis: version: v1.2 template: | 你是一个电商评论分析助手。 任务从用户评论中提取情感倾向和关键词。 输出格式严格按此JSON输出 {{ sentiment: positive | negative | neutral, keywords: [关键词1, 关键词2], confidence: 0.0 到 1.0 之间的浮点数 }} 约束 - 只提取评论中明确出现的信息。 - keywords最多5个。 用户评论{review}用YAML管理提示词的好处是非技术人员也能改。运营同学想调一下约束条件直接改YAML就行不用碰代码。4.3 接入层实现src/client/base.py定义接口src/client/openai_client.py实现具体调用。这里以兼容OpenAI接口的模型服务为例import os import time from openai import OpenAI from .base import BaseModelClient, ModelResponse class OpenAIClient(BaseModelClient): def __init__(self): self.client OpenAI( api_keyos.getenv(MODEL_API_KEY), base_urlos.getenv(MODEL_BASE_URL) ) self.model os.getenv(MODEL_NAME, gpt-3.5-turbo) def chat(self, messages: list, **kwargs) - ModelResponse: start time.time() response self.client.chat.completions.create( modelself.model, messagesmessages, temperaturekwargs.get(temperature, 0.1), max_tokenskwargs.get(max_tokens, 500) ) latency (time.time() - start) * 1000 return ModelResponse( contentresponse.choices[0].message.content, modelself.model, tokens_usedresponse.usage.total_tokens, latency_mslatency )注意temperature设成了0.1。对于信息提取类任务温度越低越好因为你需要的是稳定、可复现的输出而不是创意。温度高了同样的输入可能给出不同的情感判断这在生产环境是灾难。4.4 编排层实现src/orchestrator/review_analyzer.pyimport yaml from src.client.base import BaseModelClient class ReviewAnalyzer: def __init__(self, client: BaseModelClient, prompt_config_path: str): self.client client with open(prompt_config_path, r, encodingutf-8) as f: self.prompts yaml.safe_load(f) def analyze(self, review: str) - dict: prompt_config self.prompts[review_analysis] prompt prompt_config[template].format(reviewreview) messages [ {role: user, content: prompt} ] response self.client.chat(messages) return { raw_output: response.content, prompt_version: prompt_config[version], model: response.model, tokens_used: response.tokens_used, latency_ms: response.latency_ms }编排层只负责拼提示词、调模型、返回原始结果。它不关心输出能不能解析那是解析层的事。这样职责清晰改提示词不影响解析逻辑。4.5 解析层实现src/parser/review_parser.pyimport json import re from pydantic import BaseModel, ValidationError, field_validator class ReviewAnalysis(BaseModel): sentiment: str keywords: list[str] confidence: float field_validator(sentiment) classmethod def validate_sentiment(cls, v): allowed {positive, negative, neutral} if v not in allowed: raise ValueError(fsentiment必须是{allowed}之一) return v field_validator(confidence) classmethod def validate_confidence(cls, v): if not 0.0 v 1.0: raise ValueError(confidence必须在0到1之间) return v def parse_review_output(raw: str) - ReviewAnalysis: # 第一层直接解析 try: data json.loads(raw) return ReviewAnalysis(**data) except (json.JSONDecodeError, ValidationError): pass # 第二层正则提取 match re.search(r\{.*\}, raw, re.DOTALL) if match: try: data json.loads(match.group()) return ReviewAnalysis(**data) except (json.JSONDecodeError, ValidationError): pass raise ValueError(f解析失败: {raw[:200]})Pydantic的field_validator在这里很关键。它保证了即使模型返回了sentiment: happy这种不在枚举里的值也会被拦截。拦截后触发重试或降级而不是让脏数据流到下游。4.6 业务层组装src/service/analyze_service.pyfrom src.client.openai_client import OpenAIClient from src.orchestrator.review_analyzer import ReviewAnalyzer from src.parser.review_parser import parse_review_output, ReviewAnalysis class AnalyzeService: def __init__(self): client OpenAIClient() self.analyzer ReviewAnalyzer(client, config/prompts.yaml) def analyze_review(self, review: str) - ReviewAnalysis: result self.analyzer.analyze(review) try: return parse_review_output(result[raw_output]) except ValueError as e: # 解析失败记录日志返回降级结果 print(f解析失败: {e}, 原始输出: {result[raw_output]}) return ReviewAnalysis( sentimentneutral, keywords[], confidence0.0 )业务层很薄只做两件事调编排层拿结果调解析层转结构。解析失败时返回一个降级结果保证系统不崩。这就是“把AI当成不靠谱服务”的具体体现。4.7 跑起来看看# main.py from src.service.analyze_service import AnalyzeService service AnalyzeService() result service.analyze_review(这个手机电池太差了用半天就没电但是屏幕确实不错) print(result.model_dump())输出{ sentiment: negative, keywords: [电池差, 半天没电, 屏幕不错], confidence: 0.85 }整个流程跑通了。从输入评论到输出结构化结果中间经过了接入层、编排层、解析层、业务层。每一层职责清晰换模型只改接入层调提示词只改配置文件改输出格式只改解析层。5. 常见问题与排查技巧实录5.1 模型输出不稳定怎么办这是最常见的问题。同样的输入有时候返回JSON有时候返回一段解释文字。排查思路检查温度参数。温度高于0.3输出随机性就明显增加。信息提取类任务建议设0.1甚至0。检查提示词是否足够明确。如果提示词里说“输出JSON”但没给示例模型可能自由发挥。给一个完整的JSON示例并强调“严格按此格式”。检查是否有系统提示词。有些模型对系统提示词更敏感。把格式约束放在系统提示词里效果通常更好。5.2 解析失败率居高不下如果解析失败率超过5%不要急着加修复逻辑先去看原始输出。我遇到过几种典型情况现象原因解决输出带Markdown代码块模型习惯性包裹JSON提示词里强调“不要用代码块”字段名拼写错误模型自由发挥提示词里给完整字段名示例输出多余解释文字模型想“帮忙”提示词里加“只输出JSON不要任何解释”中文冒号导致解析失败模型用了中文标点解析层做标点归一化5.3 成本失控怎么排查成本失控通常有三个原因调用量太大、单次token太多、缓存没生效。排查步骤先看日志统计每天的调用次数和总token数。如果调用次数远超预期说明有地方在循环调用或者被恶意刷了。看单次调用的token数。如果输入token特别大说明提示词太长或者上下文塞了太多内容。精简提示词去掉不必要的示例。检查缓存命中率。如果缓存命中率低于30%说明缓存key设计有问题或者场景本身不适合缓存。实操心得我习惯在每次调用后记录一行日志包含时间戳、模型名、输入token、输出token、耗时、缓存命中与否。这些数据积累一周就能看出很多问题。不要等账单来了才去查。5.4 并发上来就报错模型服务的并发限制通常比普通API低。免费额度可能只有每分钟几次付费额度也有限制。解决方案加队列。所有模型调用走一个内部队列控制并发数。超出部分排队等待而不是直接报错。加限流。用令牌桶算法限制每秒调用次数超过就拒绝或降级。加降级。当队列满或限流触发时返回缓存结果或默认结果保证用户体验不崩。from queue import Queue from threading import Thread class ModelQueue: def __init__(self, max_workers5): self.queue Queue() self.workers [] for _ in range(max_workers): t Thread(targetself._worker, daemonTrue) t.start() self.workers.append(t) def _worker(self): while True: task, callback self.queue.get() try: result task() callback(result, None) except Exception as e: callback(None, e) finally: self.queue.task_done()5.5 模型突然返回空内容这种情况我遇到过几次通常是模型服务端的问题。排查思路先看是不是所有请求都空还是个别请求空。如果是个别可能是触发了内容过滤。检查输入内容是否包含敏感词。有些模型服务会对输入做过滤命中后返回空。检查max_tokens是否设得太小。如果设成10模型可能还没说完就被截断了。加一个空内容检测如果返回空自动重试一次。重试还空就降级。6. 效果评估与持续迭代上线不是终点6.1 怎么评估AI功能的效果AI功能的效果评估比传统功能难因为输出是自然语言没有绝对的对错。我的做法是分层评估格式正确率输出能被解析层成功解析的比例。这个指标最客观低于95%就要优化。内容准确率人工抽样检查看情感判断、关键词提取是否准确。通常抽100条算准确率。用户反馈如果有用户端加一个“这个结果有帮助吗”的反馈按钮。真实用户的反馈比人工抽样更有价值。6.2 提示词迭代的节奏提示词不是一次写好的是需要持续迭代的。我的节奏是第一周每天看解析失败日志和抽样结果快速调整提示词。第二周稳定后每周迭代一次。每次只改一个变量改完对比效果。一个月后基本稳定每月回顾一次。除非有新的边界情况否则不动。每次迭代都要记录版本号和改动内容。我通常在YAML里加一个changelog字段记录每个版本改了什么、为什么改、效果如何。6.3 什么时候该换模型换模型是个大决策不要频繁换。我通常在这几种情况下考虑换当前模型成本太高有更便宜的替代品且效果不差。当前模型在某些场景下效果明显不行换了能解决。当前模型服务不稳定经常超时或报错。换之前一定要做A/B测试。同样的输入两个模型各跑一遍对比格式正确率、内容准确率、成本、耗时。数据说话不要凭感觉。7. 一些踩过的坑和最后的建议第一个坑不要用模型做它不擅长的事。我见过有人让模型做数学计算结果错得离谱。模型擅长的是语言理解和生成不是精确计算。需要计算的地方让模型输出表达式然后用代码算。第二个坑不要忽略日志。AI应用的日志比传统应用更重要因为出问题时你很难复现。每次调用记录输入、输出、模型、耗时、token数、缓存命中。这些日志是你排查问题的唯一依据。第三个坑不要一次性上太多AI功能。我见过一个项目一周内接了五个AI功能结果每个都不稳定最后全部回滚。AI功能需要迭代一次上一个稳定了再上下一个。第四个坑不要忽视降级策略。模型服务一定会出问题区别只是早晚。提前想好降级方案返回缓存、返回默认值、走规则引擎。没有降级策略的AI应用上线就是定时炸弹。最后一个建议从最简单的场景开始。不要一上来就做多轮对话、复杂推理。先做一个文本分类或信息提取把整个工程链路跑通把缓存、重试、解析、监控都搭好。然后再逐步增加复杂度。这样每一步都稳出了问题也好定位。我自己的经验是一个稳定的简单功能比十个不稳定的复杂功能有价值得多。