1. 项目概述1.1 核心需求解析这两年AI圈子最热闹的已经从“训练一个模型”切换到“用模型做产品”。我见过太多团队卡在同一个地方模型调通了、Demo能跑了但真要上生产、接业务、扛流量立刻被Prompt飘忽不定、接口偶发超时、输出格式变来变去磨得没脾气。这个“ai-engineering-from-scratch”项目说白了就是把这些坑提前踩一遍从零开始梳理一套AI工程落地的方法论和最小实现。它不是一个训练模型的教程也不是某个框架的API手册而是讲清楚“从拿到一个模型到跑通一个稳定服务”中间那段路要怎么走。我当时的出发点很朴素团队里新同学过来与其让他们对着文档碎片拼认知不如直接给一条清晰的路径——先搞清楚AI工程的边界和分层再动手搭核心模块最后把评估、部署、监控补上。这样无论是做智能客服、内容生成还是知识库问答都能有一套可复用的底子。1.2 适用人群与前置条件这个项目适合三类人一是刚接触LLM应用开发、想建立全局视角的工程师二是已经在用LangChain之类的框架、但遇到问题只能黑盒调参的开发者三是技术管理者想评估自建AI能力需要投入多少工程成本。前置条件不高要求你写过Python、调过HTTP接口、用过Docker对Prompt这个词不陌生就行。建议先准备一台能跑Python 3.11的机器云服务器或带够了内存的Mac/Linux都行一个OpenAI或DeepSeek的API Key再加上Docker环境。整个项目全部跑通一般一周左右就够了。1.3 你最终能得到什么到项目结束时你会拥有一个完整的AI服务工程骨架包含统一模型接入层、Prompt版本管理模块、Agent工具调用循环、离线评估与回归测试集、生产环境的可观测性配置、带Basic Auth的Model Server。这些代码全部放在一个仓库里每层都有测试覆盖遇到新业务可以直接往里面加工具、加数据、加评估用例。我一直觉得AI工程最关键的价值是“降本”。不是单纯省token费用而是省团队在“修模板、调随机性、对格式”上反复挣扎的时间。这个项目就是在帮团队把这份时间省下来。2. 整体设计与技术选型2.1 架构设计思路为什么是“分层”而不是“框架”我见过不少团队一上来就铺LangChain业务逻辑和模型逻辑拧在一起出了问题根本不知道是Prompt写坏了还是链路的某一步把参数吞了。所以在设计这个项目时我坚持用一个极简的分层结构任何人都能在10分钟之内说清楚每一层是干什么的。整个系统分四层。最底下是模型接入层负责跟不同厂商的API打交道统一输入输出格式往上是数据与记忆层管知识库、向量检索、对话历史再往上是编排层负责决定“什么时候调用工具、工具返回之后怎么处理”最顶上才是业务接口层以REST API的形式把能力暴露出去。每一层只依赖下一层不跨层调用。这种设计的初衷是“可替换性”。今天用GPT-4o明天换成DeepSeek-V3或者本地部署的Qwen模型接入层改一个类就能切过去。业务层完全无感。如果当初直接堆LangChain这种替换会变成噩梦——框架上游一变你的业务代码跟着全改。提示如果你评估下来项目周期很紧、只打算跑通一个Demo可以不用这个分层。但只要是冲着生产环境去的这个分层骨架能帮你少交大量学费。2.2 技术选型对比为什么是Python FastAPI PostgreSQL先说语言。AI工程这个领域Python生态的统治地位没有争议——模型SDK、数据处理库、向量库客户端全是Python优先。Node.js在并发上有优势但AI编排层的大量类型推导和数据处理工作Python的pydantic加上类型标注明显更顺手。Web框架我选了FastAPI而不是Flask或Django原因是它原生支持异步、Pydantic校验、自动生成OpenAPI文档。后面接入流式输出时FastAPI的StreamingResponse能直接用异步生成器Flask就要自己折腾很多兼容问题。数据层用的是PostgreSQL加上pgvector插件。为什么不用专门的向量数据库因为大部分业务场景的数据量在百万级以内pgvector的召回效果和性能足够还能直接用SQL做业务字段和向量的混合过滤。少维护一个中间件对一个起步阶段的工程来说非常重要。只有当数据量真正摸到千万级以上、召回时延要求苛刻时我才会建议引入Milvus或Qdrant这类专用方案。模型服务这一层我选择自建一个轻量的Model Server而不是直接用LangServe。LangServe能把LangChain的Agent包装成API确实方便但它把太多实现细节藏了起来出了问题你只能去翻框架源码。自建的服务每一行代码都在自己仓库里调试成本长期看是更低的。模块选型选型理由Web框架FastAPI异步原生、Pydantic校验、流式输出支持好数据库PostgreSQL pgvector一套库同时管业务数据和向量部署简单模型接入自研Provider类屏蔽各家API差异支持秒级切换模型编排层自研工具循环可控性强每步都可观测、可重试部署Docker Compose开发环境一条命令拉起生产可直接扩展2.3 本地与云端部署的取舍部署策略上项目开发环境用Docker Compose一台机器“一键”启动Postgres、API服务、向量库。生产环境我没有一开始就上Kubernetes那会显著拉高复杂度。单体Docker镜像加上systemd或者云平台的容器服务完全足够支撑初期业务量。如果团队有现成的K8s环境把这个服务容器化后部署进去的成本也不高。你要注意的坑是大模型的API调用是外部IO密集型操作不像普通Web服务那样靠横向扩容就能线性提升QPS。很多时候瓶颈在APIProvider的限流上扩容只增大开销。所以设计时我特意把模型调用做成并发受限的异步模式而不是开一堆线程去硬打接口。3. 核心模块实现细节3.1 模型接入层统一Provider抽象模型接入是整个工程的地基。各家模型API的差异不只是URL和鉴权方式更麻烦的是请求参数和响应结构不统一。有的用messages数组传对话历史有的用prompt字符串有的返回choices[0].message.content有的直接给response字段。要是不做抽象业务代码会到处散落兼容逻辑。我在providers目录下定义了一个BaseProvider基类核心方法就三个chat()、embed()、stream_chat()。不同厂商各写一个子类内部做参数映射和异常转换。在config.yaml里加一个provider_type字段启动时由工厂函数加载对应实现。业务层拿到的永远是统一格式的ChatMessage对象根本不用关心背后是谁在推理。这个抽象最大的收益不是“今天换模型”而是“同一套代码同时跑多个模型”做对比评测。我会在评估模块里同时实例化两个Provider同一个Prompt分别发给不同模型效果差异一览无余对Prompt调优和模型选型判断特别有用。class BaseProvider(ABC): abstractmethod async def chat(self, messages: list[ChatMessage], **kwargs) - ChatMessage: 非流式对话 abstractmethod def stream_chat(self, messages: list[ChatMessage], **kwargs) - AsyncGenerator[str, None]: 流式对话 abstractmethod def embed(self, texts: list[str]) - list[list[float]]: 文本向量化 class OpenAIProvider(BaseProvider): def __init__(self, api_key: str, model: str gpt-4o-mini): self.client AsyncOpenAI(api_keyapi_key) self.model model async def chat(self, messages: list[ChatMessage], **kwargs) - ChatMessage: resp await self.client.chat.completions.create( modelself.model, messages[{role: m.role, content: m.content} for m in messages], **kwargs ) return ChatMessage(roleassistant, contentresp.choices[0].message.content) def stream_chat(self, messages: list[ChatMessage], **kwargs) - AsyncGenerator[str, None]: async def gen(): stream await self.client.chat.completions.create( modelself.model, messages[{role: m.role, content: m.content} for m in messages], streamTrue, **kwargs ) async for chunk in stream: delta chunk.choices[0].delta.content if delta: yield delta return gen()注意各家API的kwargs兼容性是个暗坑。OpenAI支持temperature和top_p但某些国产模型API会直接忽略不支持的参数还会对非法值抛异常。建议在Provider内部做一次白名单过滤只传该厂商明确支持的参数。3.2 Prompt工程模板化与版本管理Prompt是整个AI服务中最容易被忽略、也最容易出问题的模块。很多团队把Prompt硬编码在业务代码里测试同学改个措辞都要找开发发版本——这从根本上就是错的。Prompt应该像代码一样做版本管理、环境隔离和回归测试。我采用的是“三段式”Prompt结构系统指令、任务说明、输出格式约束。系统指令固定角色和行为边界任务说明随业务变化输出格式约束统一用JSON Schema描述。三个部分分别存放在prompts/目录下的不同模板文件里通过一个PromptManager类加载。槽位替换我用Jinja2来做而不是简单的f-string。因为实际项目中经常要对列表做循环渲染比如“参考以下历史对话”需要动态拼接多条消息Jinja2的循环语法比Python字符串拼接干净得多。输出格式约束这块我建议你务必上JSON Schema。没有约束时模型经常回你“好的以下是我整理的结果”然后才输出JSON直接导致解析失败。我在系统指令里明确要求“只输出JSON对象”并将Schema本身写进模板底部配合后端的Pydantic校验实测格式混乱率能从20%降到2%以内。你是一个专业的{{ role }}。 请在回答时严格遵守以下要求 1. 只输出Json对象不要包含多余的解释或Markdown代码块标记 2. 输出必须符合提供的Json Schema 【任务描述】 {{ task_description }} 【输出格式】 json {{ output_schema }}这里的变量role、task_description、output_schema由PromptManager在渲染时注入。把Schema也放进去模型能看到约束比在代码里“事后修正”可靠得多。 ### 3.3 Agent工具调用与循环机制 Agent自己决定什么时候调用工具这个机制没做过的人会觉得玄拆开看就是三步循环推理出意图、执行工具、把结果合并进上下文。我在项目里用一个小型循环函数实现了这个能力核心数据结构是AgentState里面保存当前上下文、已调用工具列表、剩余最大轮次。 第一步把当前状态和可用工具的JSON描述一起发给模型让模型选择“直接回答”还是“调用工具”。第二步如果模型返回了工具调用请求就在本进程内执行对应函数——比如查数据库、调外部API。第三步把工具的执行结果作为一条system消息拼回对话历史再次调用模型生成最终回复。 这一步最大的坑是循环终止条件。模型偶尔会反复调用同一个工具不肯收尾极端情况下会把你的token预算烧光。我设置了几层保护最大调用次数默认5轮、单次工具执行超时默认30秒、连续相同调用的熔断连续3次相同参数直接打断。这三道闸同时生效才能保证Agent在生产环境不“发疯”。 python async def agent_loop(user_input: str, tools: list[BaseTool], max_rounds: int 5) - str: state AgentState(messages[{role: user, content: user_input}], rounds0) while state.rounds max_rounds: response await llm.chat_with_tools(state.messages, tools) if response.tool_calls is None: return response.content for call in response.tool_calls: result await execute_tool(tools, call.name, call.arguments) state.messages.append({role: assistant, content: None, tool_calls: [call]}) state.messages.append({role: tool, tool_call_id: call.id, content: result}) state.rounds 1 return 已达最大处理轮次请简化需求后重试3.4 数据层与向量检索AI服务的数据层比传统CRUD多了一个维度不仅要存业务数据还要把非结构化文本切成向量。我在PostgreSQL里开了pgvector扩展用一张doc_embeddings表统一存储content原文、embedding向量、metadata里的业务属性比如来源、部门、时间。这样查询的时候可以在一个SQL里做向量相似度和业务条件的混合过滤非常实用。文本切块的粒度直接影响召回效果。切太大了一段文本包含多个意思检索出来虽然相关但答案不聚焦切太小了语义不完整同样影响生成质量。我跑了几组实验最稳妥的配置是块大小512个字符、重叠128个字符。这个组合既保证了语义完整性又不至于让向量维度稀释太多有效信息。向量化接口我直接调用的Embedding模型每次写库前先算好向量再入库。这里注意如果业务要求实时性你需要在写入时同步做嵌入和入库而不是用异步任务慢慢补否则数据落库了但搜不到业务方会困惑。CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE doc_embeddings ( id BIGSERIAL PRIMARY KEY, content TEXT NOT NULL, embedding vector(1536), metadata JSONB DEFAULT {}, created_at TIMESTAMPTZ DEFAULT now() ); CREATE INDEX idx_doc_embeddings_l2 ON doc_embeddings USING ivfflat (embedding vector_l2_ops);关于向量索引参数ivfflat的lists值默认是100数据量十万级以内这个够用。只有索引的probes候选列表数需要根据召回率实测调整——设小了速度快但容易漏召回设大了慢但准。一般的经验值是lists的平方根附近可以先按默认跑一轮评测再微调。4. 实操过程与关键环节拆解4.1 环境准备与依赖安装我强烈建议你先用uv或者poetry锁住Python环境不要用裸的requirements.txt。AI工程的依赖复杂度和AI模型本身一样容易失控——且LangChain类的库更新极勤昨天能跑的代码今天依赖升个级直接崩。这个项目固定的核心依赖列表如下[project] dependencies [ fastapi0.111.0, uvicorn[standard]0.30.1, pydantic2.7.4, pydantic-settings2.3.3, openai1.35.0, jinja23.1.4, pgvector0.2.4, asyncpg0.29.0, sqlalchemy[asyncio]2.0.31, httpx0.27.0, tenacity8.4.1, ]装完依赖后config.yaml里有一项必须要填embedding_model。如果用的是OpenAI的text-embedding-3-small向量维度是1536正好对应上面的建表语句。如果你换成本地部署的BGE模型维度可能变成768或1024表结构要跟着改。提示Python版本推荐3.11以上。3.10也能跑但async特性的一些行为差异会在后续调试中带来莫名其妙的问题浪费排查时间。4.2 搭建最小可运行骨架很多人的误区是上来就追求“完整功能”结果两个月过去还在写底层。我建议先把最小闭环跑通启动API服务、输入一句话、拿到模型回复。这个闭环一旦通了后面的模块都是往这个管道上挂配件。我给的骨架代码结构是app / main.py入口、app / routers / chat.py业务路由、app / providers /模型适配、app / core / config.py配置加载。启动后访问/docs就能看到自动生成的Swagger文档直接在里面测试接口。最小骨架的核心是配置管理。我把所有环境变量放在config.yaml和.env里pydantic-settings负责加载。有一个细节Key不要直接写在版本库里——.env进.gitignore模板文件统一用.env.example。团队里新同学克隆后复制一份填上自己的Key就能跑既不泄露密钥也不浪费时间。4.3 流式输出的实现与调优流式输出是AI服务的“标配体验”用户不想盯着转了5秒的菊花才看到整段结果。FastAPI的StreamingResponse天然支持异步生成器实现起来其实很直接。但流式有一个容易坑到人的点中间代理层如果开了HTTP缓冲就会把流堵住表现为“前端一直没内容等到最后一次性刷出全文”。我在app / routers / chat.py里做了一个可选参数stream默认开。核心是让Provider的stream_chat()生成器一路畅通到前端。这一路的代理——不管是Nginx还是云负载均衡——都必须关闭响应缓冲Nginx要设proxy_buffering off云LB要确认响应不缓存。还有一个细节是流式返回时的断字问题。模型是按token输出的一个中文字可能被拆成两个半截token到达前端如果直接按块渲染会出现“卡半个字”的鬼畜效果。我在服务端做了一个简单的粘合缓冲只把完整的UTF-8字符塞进流半截存下来等下一个token拼完整再发。async def event_generator(prompt: str): buf async for delta in provider.stream_chat([ChatMessage(roleuser, contentprompt)]): buf delta # 粘合半截UTF-8字符保证输出合法字符串 while buf: try: char buf.encode(utf-8).decode(utf-8) yield fdata: {json.dumps({delta: char}, ensure_asciiFalse)}\n\n buf except UnicodeDecodeError: break4.4 知识库问答的完整示例为了验证架构不是纸上谈兵我搭了一个内部知识库问答的场景把公司产品文档丢进向量库用户提问后先检索相关性最高的3段内容拼进Prompt让模型基于检索结果回答。这个概念就是RAG。在项目里我专门写了一个retrieval.py模块逻辑非常清晰。先接收用户问题做向量化然后SQL查Top-K相似文档。这里有一个进阶操作不是简单把Top-K直接丢给模型而是做一步rerank——先用向量粗筛50条再用语义精排取最相关3条。粗筛阶段我用pgvector精排用一个轻量级的cross-encoder模型。加了这步之后回答准确性提升明显但耗时也会从几百毫秒增加到1.5秒左右需要自己权衡。4.5 模型服务化与安全措施服务上线时最容易被忽视的是鉴权。AI服务就是烧token的接口裸奔出去等于给全网送福利。我在网关层加了一个简单但有效的方案Bearer Token 接口级限流。Token放在gateway模块里是一个独立于业务逻辑的中间件每个请求先校验身份再放行。限流我用的是令牌桶算法放内存里就能跑。单用户每分钟限60次单个IP每分钟限120次超出直接返回429。只是最多撑单机小规模场景如果上多副本就需要把令牌桶挪到Redis了架构上留好这个扩展点就行。5. 评估体系与可观测性5.1 评估目标为什么不能只看“感觉”AI服务上线后最大的麻烦是“不可预期”。代码你改了逻辑能测但Prompt换几个词模型回答就成了另一个风格。所以没有评估体系你根本没法判断一次改动是在变好还是变坏。我在项目里建立了一套三层评估单测断言、离线数据集评测、在线日志分析。单测断言是最便宜的那道防线。我针对核心函数写了十几个用例重点锁住输出格式必须合法、工具调用参数必须符合Schema、空输入和超长输入有兜底回复。这些断言跑一次只要几秒每次重构都先跑它能在几分钟内抓出明显回归。离线评估是核心环节。我维护了一个eval_set.jsonl字段格式是[query, expected_key_points, label]大概200条真实业务问题覆盖正常问题、边界问题和对抗样本。每次改Prompt之前我都会先跑一遍离线集记录得分改完再跑一遍对比差异。有数据说话就不会出现“感觉好多了但不知道好在哪”的局面。5.2 评估指标的选择逻辑评估指标选几个是够用的我用了三个准确率、召回关键信息率和格式合规率。准确率就是LLM-as-Judge拿GPT-4o当裁判把系统回答和标准答案一起发过去打分按1到5分输出最终算平均分。这个方式便宜、快、自动化缺点是模型自身偏差会带来噪声所以只能做横向相对比较不能当绝对真理。召回关键信息率是人工抽检的一个指标看回答里是否包含正确答案中出现的实体和数字格式合规率直接由解析器判断只要是输出JSON的功能这个指标能自动化统计非常有效。提示用LLM当裁判时最好固定Judge用的模型版本和temperature。如果今天用GPT-4o明天用DeepSeek得分尺度完全不同评测结果就失去可比性。术语上叫“评测方差”实际上就是说你的评测工具本身不稳定。5.3 日志架构从请求到生成的完整链路追踪生产环境的AI服务调试最怕的是“用户说回答有问题但你不知道他当时发了什么”。我把日志设计成每个请求一个request_id从入口中间件开始生成贯穿向量检索、模型调用、工具执行全链路。在用日志排查问题时我只需按request_id筛一遍整条链路的耗时和中间结果都排着队等你审。日志不能只记输入输出。我在模型调用层额外记录了Prompt的版本号、temperature参数、模型返回的原始内容、本次调用的token消耗。这几个字段排在一起就能回答运维同学最常问的两个问题——“为什么这次回答质量这么差”和“为什么这个月成本涨了这么多”。对话历史的落库也在这一层完成。我建了conversation_logs表用户ID、请求ID、输入输出、耗时、token数全查得到。合规要求也很重要聊天记录留痕是基本功。5.4 在线监控指标与告警生产环境的三大核心指标调用成功率、P95时延、Token消耗速率。任何一个异常都直接关系用户体验和钱。我在monitoring / metrics.py里封装了三个Prometheus指标类型Counter记录成功失败次数、Histogram记录时延分布、Gauge记录当前并发数。API层通过中间件自动埋点不需要业务代码额外关心。告警规则也预设了两条成功率低于95%持续2分钟触发页面告警P95时延超过5秒持续5分钟触发值班电话。这两条规则极为粗糙但能保证初期不出大事。上线后第一个周末我几乎泡在DashBoard前看曲线。前三天波动大是正常的等业务量稳定了再根据真实分布去调告警阈值。不要第一天就把阈值定得很灵敏否则告警轰炸会让团队麻木最后一出事都没人响应。6. 常见问题与排查实操6.1 问题排查速查表从零开始搭AI工程的过程本质就是一个不停填坑的过程。下面这张表是我被现实教育过后沉淀出来的高频问题清单适合打印出来贴在工位上。现象可能原因排查手段解决方案输出偶尔多出Markdown符号Prompt没约束死格式打开log模板原文看注入后全文模板底部追加JSON Schema前端再兜底清洗调用超时频繁模型API限流或网络不稳看Provider日志里HTTP状态码加tenacity重试指数退避降并发数向量召回结果不相关切块粒度过大或Embedding模型不匹配抽样打印召回文本对比调块大小和重叠区间换Embedding模型重灌库Agent反复调用同一工具上下文里结果不明确或工具描述有歧义打开工具调用的追踪日志改写工具描述加入相同调用熔断Docker里连不上宿主机Postgres容器网络配置问题docker compose ps看端口映射检查ports配置改用host.docker.internal请求时快时慢共享大模型账号限流监控不同实例的时延分布增加专属额度或多Key轮询6.2 典型排查实录一个“加深思”的陷阱有一次业务反馈搜索问答质量下降用户问“如何部署”系统回答了一堆部署细节但没提初始化参数。我第一反应是向量检索召回出了问题但排查日志后发现“标题过滤”条件加错了——新上线的功能给检索加了一个过滤条件想过滤掉过期的文档结果因为日期字段格式不匹配把几乎所有文档都过滤掉了。召回结果只剩两三段无关内容模型只能硬着头皮答非所问。这个案例给我的教训是排查问题不要老盯着模型很多“AI问题”本质是工程问题。80%的情况下先看数据链路、过滤条件、请求参数都比怀疑模型要强得多。6.3 成本控制与Token优化成本问题是AI工程里最现实的焦虑。我做过的优化手段按性价比排序如下。最优先级是加语义缓存相同或高度相似的问题直接返回缓存结果能省大约30%的重复调用。我用的方案是在Redis里存向量查询前先做一次判决相似度高于0.95直接用历史答案。第二个手段是Prompt瘦身。把Prompt从1200字压到800字生成质量不会差太多但输入Token减少意味着成本直接下降三分之一。这一步需要配合离线评估集来验证不能凭感觉乱砍。第三个手段是模型分级简单的意图识别用便宜的小模型复杂的写作和推理才调用大模型。全站统一用大模型预算再厚也顶不住用户量上来。注意稳健的成本优化核心原则是——先保住质量基线再谈省钱。每次改动都要拿评估集跑一遍对比最好把token消耗也计入评估报告这样优化才有数据支撑。7. 项目复盘与经验沉淀这个工程做下来最大的收获不是代码量而是让我想清楚了一件事AI工程这个方向的复杂度从来不在模型本身而在“如何把不可预测的东西封装成可预测的服务”。你在传统后端里学到的分层、抽象、测试、监控到了AI领域不但没有过时反而比以前更重要。如果让我重新做一遍我会在第一天就把eval_set建起来而不是等问题出现了再去补测试数据。好的测试集是这个项目的锚锚稳了后面怎么重构都不慌。我还会更早地引入request_id链路追踪而不是等出现了好几例“问题复现不了”的尴尬局面才补上。最后再分享一个小技巧给你的每个Prompt模板打上版本号然后在日志里记下当前版本。当你某天把Prompt从V1改到V8翻日志时就会发现——所谓模型变笨了很多时候根本就是从V3升到V4那一次措辞调整引入的。有了版本标记回归定位就是看一条日志的事。按这个思路去做你的AI工程就能越用越稳。