AI工程落地指南：从零搭建稳定可靠的LLM应用服务

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，确实方便，但它把太多实现细节藏了起来，出了问题你只能去翻框架源码。自建的服务每一行代码都在自己仓库里，调试成本长期看是更低的。

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_key=api_key) self.model = model async def chat(self, messages: list[ChatMessage], **kwargs) -> ChatMessage: resp = await self.client.chat.completions.create( model=self.model, messages=[{"role": m.role, "content": m.content} for m in messages], **kwargs ) return ChatMessage(role="assistant", content=resp.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( model=self.model, messages=[{"role": m.role, "content": m.content} for m in messages], stream=True, **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}], rounds=0) 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 = [ "fastapi==0.111.0", "uvicorn[standard]==0.30.1", "pydantic==2.7.4", "pydantic-settings==2.3.3", "openai==1.35.0", "jinja2==3.1.4", "pgvector==0.2.4", "asyncpg==0.29.0", "sqlalchemy[asyncio]==2.0.31", "httpx==0.27.0", "tenacity==8.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(role="user", content=prompt)]): buf += delta # 粘合半截UTF-8字符，保证输出合法字符串 while buf: try: char = buf.encode("utf-8").decode("utf-8") yield f"data: {json.dumps({'delta': char}, ensure_ascii=False)}\n\n" buf = "" except UnicodeDecodeError: break

4.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工程的过程，本质就是一个不停填坑的过程。下面这张表，是我被现实教育过后沉淀出来的高频问题清单，适合打印出来贴在工位上。

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工程就能越用越稳。