从零动手搭建AI工程全链路：模型统一接入、RAG检索、Agent编排与评估上线

做AI工程和写AI Demo是完全不同的两件事。很多人拿着OpenAI的SDK调通一个聊天接口就觉得已经会做AI工程了但真正把功能推上线、接上业务、扛住流量、持续迭代的时候才会发现坑有多深。这个项目叫ai-engineering-from-scratch本质就是丢掉那些开箱即用的框架从零开始搭建一套完整的AI应用工程体系从模型调用、提示词管理、知识库检索到Agent编排、服务部署、效果评估全部自己动手实现一遍。这篇文章我会把整个项目的设计思路、关键模块、实操过程以及我踩过的坑都拆开讲清楚适合那些已经能调通模型接口、但想真正理解AI工程化全流程的开发者。1. 项目核心思路从零构建AI工程到底在构建什么1.1 AI工程不只是一个模型调用服务先定义边界。市面上很多所谓的AI项目本质上是把一个模型API包了一层HTTP接口再加上鉴权和计费就敢叫AI中台。但这个项目我给自己定的标准是不只要能用还要可控、可测、可运维。所谓AI工程至少要覆盖四层内容模型层多个模型统一接入、路由切换、参数透传、Budget控制数据层文档加载、切分、向量化、存储、检索也就是RAG链路逻辑层Agent规划、工具调用、记忆管理、多轮对话状态维护工程层可观测性、评估体系、灰度发布、配额管理、成本核算。为什么要从零写而不是直接上LangChain我的体会是框架帮你解决80%的常规场景但剩下20%的细节问题——比如prompt模板的版本管理、不同模型的输出格式差异、工具调用解析失败的兜底逻辑——恰恰是线上故障的头号来源。这些只有自己手写过一遍才知道坑在哪里。1.2 项目边界与技术栈选择作为个人项目不可能什么都自己做所以要明确哪些自研、哪些复用。我的方案是模型层自己封装统一接口底层接入多个模型服务商但业务代码只面对一套抽象向量数据库直接用开源的pgvector或Qdrant不重复造轮子Agent框架不引入重量级编排工具用代码显式定义状态机避免黑盒调度评估体系自建因为当时评估平台还不成熟而且技术栈是Python自己写一套回归测试成本并不高。技术栈我选了Python FastAPI PostgreSQLpgvector Redis前面都是主流选择没什么可纠结的。真正需要考虑的反而是模型网关的设计这个我单独拿出来讲。为什么这样选Python生态在AI领域积累最深FastAPI的异步性能够用而且写起来直观pgvector能让我少维护一套专用向量库Redis既做缓存又做会话存储。全套下来单机也能跑起来方便迭代。2. 核心模块设计与方案取舍2.1 模型网关统一接入层是AI工程的骨架模型网关是整个系统最容易被低估的部分。没有网关业务代码直接调各家SDK等到要换模型或加新模型的时候改动量会非常痛苦。我的网关抽象了三个核心能力统一模型接口所有下游只调用一个chat()方法传入model_name、messages和参数网关负责路由。模型参数归一不同模型的temperature、top_p、max_tokens语义基本相同但默认值和边界不同网关统一校验并转换。优雅降级某个供应商限流或返回异常时网关按预设策略切换到备用模型而不是直接抛错给用户。这里有个关键设计降级不能静默。每次降级都要记录到日志和监控否则线上出了问题你根本不知道用户实际用的是哪个模型排查起来会非常被动。另外流式输出必须在一开始就支持。非流式接口写起来简单但真实用户体验差距太大后面再改会造成全链路的回调重构。我第一版就上了SSEServer-Sent Events协议后面扩展流式输出时几乎没动业务代码。2.2 提示词管理把Prompt当代码管起来Prompt在工程里的地位远比很多人想的更重要。它不是一个字符串而是会频繁变动的配置代码。我见过太多项目把Prompt散落在代码里想改一个词要全局搜索完全没有版本管理。这个项目的做法是每个Prompt独立文件存放文件名即唯一ID用元信息标注适用模型因为Claude和GPT对格式的敏感度不同、版本号、变更说明服务启动时统一加载到内存运行时通过ID引用支持动态刷新。为什么要这样做因为线上Prompt不能直接改。我测试过很多次一个看似无害的措辞调整可能在边界case上让输出格式直接崩掉必须有回滚能力。代码化管理之后Prompt变更可以走普通的代码审查、测试、发布流程风险大大降低。2.3 RAG链路切分策略比向量模型更影响效果做RAG的人容易把注意力放在embedding模型选型上但实际调试下来文本切分对回答质量的影响往往更大。切得太碎上下文碎片化切得太整检索噪音增多。我的经验是按文档结构切分而不是按固定字符数切分。具体实现分几步读取文档时先识别标题层级markdown的#、PDF的章节把同一章节的内容聚合在一起如果某一块超过阈值比如800 token再按段落或句子边界二次切分每条切分结果记录原始文档ID、章节路径和上下文索引方便召回后拼回完整段落。其次是检索策略。只用向量检索容易漏掉精确匹配所以我加了混合检索向量分数和关键词分数加权合并权重可以通过配置调整。实测下来针对技术文档类场景关键词权重给到0.3到0.4效果比较稳定纯语义场景就回到0.1以下。2.4 Agent工作流用状态机代替黑盒编排Agent是AI工程里最容易写出看起来能跑、实际不稳定代码的部分。很多框架把规划、执行、反思包装成黑盒出问题根本不知道是哪一步挂了。所以我坚持用显式状态机来定义工作流。我的状态定义是idle接收用户意图planning生成执行计划决定调用哪些工具及顺序tool_execution逐个执行工具调用并收集结果synthesizing基于工具结果生成最终答复fallback当计划失败或工具异常时进入兜底对话。每个状态流转都记录trace_id和完整上下文方便事后回放。有人会问这样是不是比用LangChain的AgentExcutor笨确实笨但好处是每个环节都透明可控。生产环境里一个能清晰回答为什么走了这一步的Agent比一个聪明但不可解释的Agent值钱得多。2.5 评估体系没有评估AI工程寸步难行评估是AI工程里最容易被砍掉、但最不该砍的部分。因为LLM输出有随机性改动一个prompt你无法靠肉眼看一下判断整体效果是否退化。我搭了一套轻量级评估流程离线评估集几百条真实用户问题和对应的期望行为标签覆盖正常问答、拒绝回答、工具调用、多轮追问等场景自动化评测用LLM作为裁判按检索相关性、回答准确度、格式规范性、安全性几个维度打分回归门禁每次改动prompt或检索逻辑都必须跑一遍评估集得分低于基线则不允许合并。这套流程不复杂但非常有效。它把AI应用开发从改了试试看变成了改了跑分看至少能挡住大部分明显的回退。3. 实操过程核心环节的完整实现思路3.1 模型网关代码骨架说太多理论不如直接看代码。模型网关的核心就是一个异步函数加上路由表和朋友错误处理。结构大致如下class ModelGateway: def __init__(self, providers: dict, fallback_policy: dict): self._providers providers self._fallback_policy fallback_policy async def chat(self, request: ChatRequest): provider_name self._route(request.model) provider self._providers[provider_name] try: return await provider.complete(request) except ProviderRateLimitError: backup self._fallback_policy.get(request.model) if backup: logger.warning(model %s rate limited, fallback to %s, request.model, backup) return await self._providers[backup].complete(request) raise很朴素的代码但有几个细节容易被忽略ProviderRateLimitError和普通错误要分开捕获因为应对策略不同降级后要打日志并且日志里带上request_id方便追踪超时时间不能设成相同的不同模型的响应速度差异可能很大统一超时会导致部分模型频繁误判超时。还有一个很实用的技巧给每个请求加一个max_retries但重试只在幂等场景用。对话生成不是天然幂等的用户可能因为重试看到重复回复所以我的默认策略是不重试只降级。3.2 RAG链路的最小实现RAG看起来复杂核心链路其实只有三步加载、切分、检索引擎。我用比较精简的方式实现了一遍。文档加载和切分的伪代码def split_document(doc: Document) - list[Chunk]: sections split_by_heading(doc.content) chunks [] for section in sections: if estimate_tokens(section.text) MAX_CHUNK_TOKENS: for sub_section in split_by_paragraph(section.text): chunks.append(Chunk(doc_iddoc.id, textsub_section, meta...)) else: chunks.append(Chunk(doc_iddoc.id, textsection.text, meta...)) return chunks检索时我没有直接只用向量检索而是做了一个简单的混合召回def hybrid_search(query: str, top_k: int 5): vec_results vector_search(query, top_k * 2) kw_results keyword_search(query, top_k) return merge_rank(vec_results, kw_results, keyword_weight0.3)merge_rank里用的分数归一化要注意向量距离和关键词分数的量纲完全不同不做归一化直接加权等于没加权数。我常用的做法是把向量相似度映射到0到1区间然后按权重线性合并。3.3 为Agent接入工具调用Agent要落地工具调用是绕不开的。我实现了一套轻量级的函数声明即工具机制让Agent能调用搜索、数据库查询、计算器等业务函数。步骤是这样的每个工具定义成Python函数用装饰器描述它的参数schema系统Prompt中注入所有工具的JSON Schema描述模型返回的tool_call对象解析后从工具注册表找到对应函数并执行执行结果返回给模型让模型继续推进对话。我踩过一个很痛的坑某些模型的工具调用格式不稳定偶尔会合并参数或者漏掉参数名。所以我在解析层加了Schema校验解析失败时不是盲目重试而是把错误信息回传给模型让模型自己修正调用参数。这个方法极其有效实测工具调用成功率从85%左右提升到98%以上。3.4 上下文管理窗口再大也不能无脑塞很多人以为上下文窗口够大就能多塞内容但实际模型在超长上下文里的表现会退化而且Token成本是硬约束。我的上下文管理策略是多轮对话中保留系统指令、最近N轮完整对话、当前任务相关检索结果中间老对话做摘要压缩存到Redis每条检索结果先计算与问题的相关性低于阈值的不进入上下文而不是无脑Top K全塞。这里有个关键参数保留最近轮数N我试过多组值最终在10到15轮之间效果比较平衡。太短会让模型失去前文信息太长则既浪费Token又可能引入噪音。压缩摘要的触发性策略是当对话历史超过实际Token上限的60%时启动避免每次请求都做无谓的摘要计算。4. 生产落地部署、监控与持续迭代4.1 服务部署与扩展性考量整套系统我用Docker Compose编排分为api服务、worker服务、PostgreSQLpgvector、Redis四个容器。API负责对话和流式响应worker处理文档入库、批量检索索引、异步评估这些重活。部署时需要注意一个很多人忽略的问题嵌入模型的内存占用。我用的embedding模型虽然不大但加载到GPU后再加载多个副本就很浪费。解决方案是单独起一个embedding服务供API和worker共用而不是每个进程各加载一份否则16G内存看着很大几个进程就吃完了。另外flow式响应要求代理层和客户端之间的连接不能随意超时。我在Nginx层把proxy_read_timeout设为300秒同时启用HTTP/2支持全链路保持长连接否则浏览器侧容易在SSE传送过程中断流。4.2 可观测性日志、指标、Trace一条都不能少AI应用的可观测性比传统Web服务更复杂因为你不只要看系统指标还要看模型输入的提示词和输出内容。我的做法是每个请求生成全局trace_id从API入口一直传到模型网关、检索器、Agent状态机结构化日志记录每个环节的耗时、Token消耗、模型名称、Retry次数单独建了一张llm_log表保存每次请求的输入消息、输出内容和评分用于离线分析关键指标上报到Prometheus包括QPS、平均首字延迟、工具调用成功率、Token消耗速率。首字延迟Time to First Token是我最关注的指标之一。模型流式输出的首字延迟直接决定用户第一感知如果超过3秒即使整体响应很快用户也会觉得卡。我在网关里单独统计了这个数字设置了告警。4.3 持续评估与回归测试的工程化评估如果只在本地手动跑时间一长就会荒废。我把评估集和评估脚本接入了CI/CD流程每次代码合并前自动运行一轮AI回归测试输出一份对比报告。报告里我会关注几类变化整体得分是否低于上一版本哪些用例得分下降超过10分是否有新增的工具调用失败案例。这套机制坚持了三个月后积累了非常宝贵的经验教训。有一次我把检索的keyword_weight从0.2调到0.4本地试了几个问答感觉更好但回归测试显示法律类问题得分下降超过15%因为这类问题的答案依赖精确法条而不是关键词我差点把一个糟糕的参数上线。5. 常见问题速查与避坑指南5.1 高频问题排查表现象可能原因解决方案流式响应中途断开代理层超时或缓冲未关关闭代理缓冲调大read_timeout工具调用不断重试死循环工具返回错误被模型反复看到限制最大工具调用轮次为3超限强制fallback检索结果与问题完全无关向量化没考虑同义词改写增加关键词召回并加权或用查询改写模型输出格式不符合JSONPrompt缺少示例或temperature过高加few-shot示例temperature降到0.2以下多轮对话后回答质量明显下降上下文中的历史轮次太多启用摘要压缩保留最近10-15轮5.2 几个值得单独说的深度教训第一模型输出的JSON解析不能用裸json.loads。模型经常输出带Markdown代码块包裹的JSON或者输出里混入额外字段。我的解析层先做格式清洗再走Schema校验清洗函数大概二十行但避免了大量线上解析报错。第二缓存不能只看命中率。我做语义缓存时发现命中率高达30%但用户反馈响应没变化。原因是缓存命中了旧的向量检索结果而底层知识库已经更新了。后来我给缓存键加了知识库版本号彻底解决。第三不要盲信LLM-as-a-Judge的分数。它适合做趋势判断不适合当绝对标准。我评估时至少用两个模型做交叉评分当分数差异过大时丢弃该样本并人工复核避免单个模型的偏好在回归门禁里引入噪声。6. 经验总结与下一步扩展思路6.1 我在实操中最深刻的体会做这个项目最大的收获不是代码量而是建立了对AI系统稳定性的正确认知。LLM本身就是概率系统不具备传统软件那种确定性所以AI工程的核心工作其实是把不确定性控制在业务可接受的范围内。方法无非三样让每一步都可见、让影响范围可缩减甚至可回滚、让退化能被及时发现并归因。只要这三件事做到位哪怕模型不完美系统也能稳稳当当跑下去。6.2 这个项目还能怎么扩展后面如果要继续迭代我比较想补三个方向更精细的Agent记忆机制目前是简单摘要下一步按实体和事件结构化管理在线评测系统把离线回归扩展到线上灰度流量对比模型成本预算控制实现按用户的配额和模型优先级自动路由。最后分享一个小技巧所有关键路径上的模型调用一定要打印token耗时和费用。很多项目上线后才发现模型成本超预算就是因为没有最基础的成本观测。这个指标最好在网关层统一统计不要等到业务层去加。