尧图网页设计 
1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、面试、做项目咨询时反复遇到的一个痛点太多人把“会调API”当成了“会做AI工程”。我见过不少简历上写着“熟悉大模型应用开发”的同学聊下来发现他们的全部经验就是pip install openai然后写个while循环拼prompt。一旦线上QPS上来、上下文变长、多轮对话状态需要管理、模型输出格式不稳定、成本失控立刻就懵了。这不是他们的错因为市面上绝大多数教程都在教“怎么用”很少有人系统性地讲“怎么从零把一套AI工程体系搭起来”。所以当我看到“ai-engineering-from-scratch”这个项目标题时我脑子里立刻浮现出一个清晰的画像这是一个面向有一定编程基础、但希望真正理解AI工程全貌的开发者从最底层开始一步步构建出可运行、可扩展、可维护的AI应用系统的实践项目。它要解决的核心问题不是“怎么调用某个模型”而是“当你要把AI能力变成一个真正的产品功能时中间到底需要哪些环节每个环节的关键决策是什么”。这篇文章我就围绕这个项目标题把我自己从零搭建AI工程体系的经验、踩过的坑、以及那些文档里不会写的细节完整地拆解一遍。适合谁看如果你已经会写Python用过至少一个AI模型的API但总觉得自己的项目“能跑但不敢上线”那这篇内容就是为你准备的。如果你是完全的新手也没关系我会把每个环节的“为什么”讲清楚让你知道每一步在解决什么问题。2. 整体架构设计先想清楚数据怎么流再动手写代码2.1 为什么我不建议从模型选型开始很多人做AI项目第一反应是“我该用哪个模型”。GPT-4还是Claude开源模型能不能打要不要微调这些问题当然重要但它们不是起点。真正的起点是你的数据怎么流。我举个具体的例子。假设你要做一个“智能客服助手”能根据用户问题从知识库检索答案并生成回复。如果你一上来就纠结用哪个模型你可能会花三天时间对比各种模型的benchmark最后选了一个“最强”的。但真正上线后你会发现瓶颈根本不在模型而在检索环节——知识库的切分策略、向量化的质量、召回率的高低这些才是决定用户体验的关键。模型再强检索回来的内容不对生成的结果就是胡编乱造。所以“ai-engineering-from-scratch”这个项目我建议的架构设计顺序是先定义数据流再确定组件边界最后才做技术选型。数据流是什么就是用户输入进来之后经过哪些处理步骤最终输出什么结果。每个步骤需要什么数据、产生什么数据、数据量级大概多少、延迟要求是多少。把这些想清楚技术选型就是水到渠成的事。2.2 一个可落地的分层架构参考基于我自己的实践经验一个从零搭建的AI工程体系可以分成四层。这个分层不是理论模型是我在实际项目中反复调整后觉得最顺手的结构。接入层负责接收用户请求做初步的参数校验和限流。这一层不需要太复杂一个FastAPI或者Flask应用就够了。关键是做好请求日志把每次请求的输入、输出、耗时、token消耗都记下来。这些日志后面做成本分析和效果优化时就是金矿。编排层这是整个系统的核心。它负责决定“这次请求要走哪条路径”。比如用户问的是事实性问题走检索增强生成RAG路径用户问的是需要计算的问题走工具调用路径用户只是闲聊走直接生成路径。编排层还需要管理多轮对话的状态决定哪些历史信息要带入下一轮。这一层我通常用一个状态机或者简单的规则引擎来实现不一定非要上LangChain那种重框架。能力层这一层才是模型、检索器、工具函数等具体能力的实现。模型调用要做统一的封装包括重试、降级、超时控制。检索器要支持多种索引方式比如向量检索、关键词检索、混合检索。工具函数要定义清晰的输入输出schema方便编排层调用。数据层包括向量数据库、关系型数据库、缓存、对象存储等。这一层的关键是数据的一致性和可追溯性。比如每次检索用到的知识库版本要记录每次模型调用的参数要留存方便后面做A/B测试和问题排查。注意这个分层不是绝对的。小项目可以合并大项目可以再拆。但核心思想是每一层只做自己该做的事层与层之间通过清晰的接口通信。这样后面任何一层要换实现都不会影响其他层。2.3 技术选型背后的取舍逻辑有了架构分层技术选型就有章可循了。我拿几个关键决策点来说说我的取舍逻辑。模型调用方式直接用官方SDK还是自己封装HTTP请求我的建议是初期用官方SDK快速验证但一定要在SDK外面再包一层自己的接口。为什么因为你需要统一处理重试、超时、降级、日志。如果直接散落在业务代码里调SDK后面想换模型供应商或者加个缓存就得改遍所有地方。我自己的做法是定义一个ModelClient抽象类所有模型调用都通过它具体实现可以是OpenAI、Anthropic或者本地模型。向量数据库选型Chroma、Qdrant、Milvus、PgVector选哪个我的经验是数据量在百万级以下、团队没有专职运维的情况下PgVector是最省心的选择。它直接跑在PostgreSQL里你不需要额外维护一个数据库服务备份、监控、权限管理都能复用现有的PostgreSQL工具链。数据量上亿了再考虑Milvus或者Qdrant。Chroma适合本地开发和小规模原型但生产环境我不太推荐主要是持久化和并发方面需要额外考虑。编排框架LangChain、LlamaIndex还是自己写我的观点可能有点反主流如果你是要做一个长期维护的产品我建议自己写编排逻辑。LangChain的抽象层次太多出了问题很难调试而且版本更新频繁升级一次可能改一堆代码。自己写编排逻辑代码量其实不大但可控性完全不一样。当然如果你只是做原型验证LangChain能帮你省不少时间。部署方式容器化是必须的。Docker Compose适合单机部署Kubernetes适合多服务、需要弹性伸缩的场景。我的建议是即使初期只用一台服务器也要用Docker Compose把各个服务拆开。这样后面要迁移到K8s或者加服务成本会低很多。3. 核心模块拆解每个环节的坑我都替你踩过了3.1 模型调用封装别让重试逻辑散落各处模型调用看起来简单不就是发个HTTP请求吗但实际生产环境中你需要处理的情况远比想象中多。网络抖动、API限流、模型返回格式错误、超时、余额不足每一种情况都需要不同的处理策略。我自己的ModelClient封装通常包含这几个关键设计。第一统一的重试策略。对于网络错误和限流错误用指数退避重试对于参数错误和余额不足直接失败不重试。第二超时控制。每个请求设置一个合理的超时时间比如30秒超时后走降级逻辑。第三降级策略。主模型不可用时自动切换到备用模型或者返回一个兜底回复。第四日志记录。每次调用的输入、输出、耗时、token消耗、是否重试、最终状态全部结构化记录。这里有个细节很多人会忽略token消耗的计算。不同模型的token计算方式不一样而且流式输出和非流式输出的token统计也有差异。我建议在封装层统一做token估算用tiktoken这样的库虽然不完全准确但足够做成本监控了。精确的token消耗可以从API返回的usage字段获取但流式输出时很多API不返回usage这时候就需要自己估算。# 一个简化的ModelClient封装示例 class ModelClient: def __init__(self, primary_model, fallback_modelNone): self.primary primary_model self.fallback fallback_model self.max_retries 3 def call(self, messages, **kwargs): for attempt in range(self.max_retries): try: response self._call_model(self.primary, messages, **kwargs) self._log_success(response) return response except RateLimitError: time.sleep(2 ** attempt) except (NetworkError, TimeoutError): time.sleep(1) except ModelError as e: self._log_error(e) break if self.fallback: return self._call_model(self.fallback, messages, **kwargs) raise ModelUnavailableError(All models failed)实操心得重试次数不要设太多3次足够了。我见过有人设10次重试结果一个请求卡了半分钟用户体验极差。另外重试的时候要考虑幂等性如果模型调用有副作用比如写数据库重试前要确认是否已经执行过。3.2 检索增强生成切分策略决定成败RAG是现在AI工程里最常用的模式但也是最容易做砸的环节。我见过太多项目模型选的是最好的prompt写得也很用心但检索回来的内容驴唇不对马嘴生成的结果自然好不了。RAG的核心在于检索质量而检索质量的第一道关卡是文档切分。切分策略没有万能公式但有几个原则可以参考。第一按语义切分不要按固定字数切分。一个完整的段落、一个章节、一个问答对都是自然的语义单元。按固定字数切分很容易把一句话切成两半检索的时候两边都匹配不上。第二切分后的块要有适当的重叠。比如每个块500字相邻块之间重叠100字这样跨块的信息不会丢失。第三每个块要保留足够的元数据。来源文档、章节标题、页码、时间戳这些信息在生成阶段可以用来做引用和过滤。向量化模型的选择也很关键。我的经验是中文场景下BGE系列或者M3E系列的效果通常比OpenAI的text-embedding-ada-002好。而且这些模型可以本地部署没有API调用成本数据也不用出本地。当然如果你的数据量特别大本地部署的推理速度可能是个瓶颈这时候可以考虑用GPU加速或者用API。检索策略上我强烈建议用混合检索向量检索加关键词检索然后做融合排序。纯向量检索对语义匹配好但对精确匹配比如产品型号、人名容易漏。关键词检索比如BM25对精确匹配好但对语义变体不行。两者结合召回率能提升不少。融合排序可以用RRFReciprocal Rank Fusion简单有效不需要训练。# RRF融合排序的简化实现 def rrf_fusion(vector_results, keyword_results, k60): scores {} for rank, doc_id in enumerate(vector_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, doc_id in enumerate(keyword_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)注意检索回来的内容不要一股脑全塞给模型。上下文长度是有限的而且塞太多无关内容反而会干扰模型。我的做法是先做一轮粗排取top 20然后用一个小的交叉编码器做精排取top 5送给模型。交叉编码器可以用BGE-reranker效果很明显。3.3 对话状态管理多轮对话不是简单拼历史多轮对话是AI应用里最复杂的部分之一。很多人以为多轮对话就是把历史消息拼在一起发给模型但实际做起来会发现一堆问题历史太长超出上下文限制、用户切换话题后旧信息干扰、指代消解“它”、“那个”指什么需要额外处理。我的做法是把对话状态分成几个部分来管理。短期记忆最近几轮对话的原始消息直接拼在prompt里。长期记忆从历史对话中提取的关键信息比如用户的名字、偏好、之前提到的事实用结构化的方式存储需要时再注入prompt。会话摘要当对话轮次超过一定数量时把早期的对话压缩成摘要替代原始消息节省上下文空间。指代消解是个容易被忽略的问题。用户说“帮我查一下它的价格”这个“它”指的是上一轮提到的商品。如果直接把这句话发给模型模型可能不知道“它”是什么。我的做法是在编排层加一个轻量的指代消解步骤用规则或者小模型把指代词替换成具体实体再发给主模型。这个步骤不需要太精确但能显著提升多轮对话的准确率。会话状态的存储我建议用Redis或者内存数据库设置合理的过期时间。不要存在关系型数据库里因为读写太频繁而且状态数据通常不需要长期持久化。如果需要做对话分析可以异步把状态快照写到数据仓库。3.4 输出解析与校验别相信模型会乖乖返回JSON如果你让模型返回JSON格式十次里可能有八次是合法的但剩下两次会给你各种惊喜多一个逗号、少一个引号、用单引号、嵌套结构不对、甚至直接返回一段解释文字。在生产环境里这种不确定性是致命的。我的做法是三层防护。第一层在prompt里明确要求返回JSON并给出schema示例。第二层用JSON解析器尝试解析如果失败尝试修复常见错误比如去掉尾随逗号、替换单引号。第三层如果还是失败用一个小的修复模型或者规则引擎把模型的输出重新格式化成合法JSON。如果三层都失败走降级逻辑返回一个默认值或者错误提示。对于关键字段还要做业务校验。比如模型返回了一个日期你要检查这个日期是否合法、是否在合理范围内。模型返回了一个金额你要检查是否为正数、是否超出限额。这些校验看起来琐碎但能避免很多线上事故。# 输出解析的三层防护示例 def parse_model_output(raw_output, schema): # 第一层直接解析 try: data json.loads(raw_output) return validate_schema(data, schema) except json.JSONDecodeError: pass # 第二层修复常见错误 try: fixed fix_common_json_errors(raw_output) data json.loads(fixed) return validate_schema(data, schema) except (json.JSONDecodeError, ValidationError): pass # 第三层用规则提取关键字段 try: data extract_fields_by_regex(raw_output, schema) return validate_schema(data, schema) except ExtractionError: raise OutputParseError(Failed to parse model output)实操心得在prompt里给schema示例的时候尽量用具体的值不要用占位符。比如写{name: 张三, age: 30}比写{name: string, age: number}效果好得多。模型对具体示例的遵循度更高。4. 实操全流程从零到一搭建一个可运行的AI工程原型4.1 环境准备与项目初始化说了这么多设计层面的东西现在来点实际的。我带你走一遍从零搭建一个AI工程原型的完整流程。这个原型是一个“技术文档问答助手”用户可以提问系统从技术文档中检索相关内容并生成回答。首先是环境准备。Python 3.10以上推荐用uv或者poetry管理依赖比pip快很多。核心依赖包括FastAPI做Web框架OpenAI或者Anthropic的SDK做模型调用PgVector做向量存储tiktoken做token计算pydantic做数据校验。# 用uv初始化项目 uv init ai-engineering-from-scratch cd ai-engineering-from-scratch uv add fastapi uvicorn openai psycopg2-binary pgvector tiktoken pydantic python-dotenv项目结构我建议这样组织ai-engineering-from-scratch/ ├── app/ │ ├── main.py # FastAPI入口 │ ├── config.py # 配置管理 │ ├── models/ # 数据模型 │ ├── clients/ # 模型客户端封装 │ ├── retrieval/ # 检索模块 │ ├── orchestration/ # 编排逻辑 │ └── utils/ # 工具函数 ├── scripts/ │ ├── ingest.py # 文档入库脚本 │ └── evaluate.py # 效果评估脚本 ├── tests/ ├── docker-compose.yml └── .env配置管理用pydantic的BaseSettings把模型API密钥、数据库连接串、各种超时参数都放在环境变量里。不要硬编码在代码里也不要提交到git。# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str database_url: str model_name: str gpt-4o-mini embedding_model: str text-embedding-3-small max_context_tokens: int 4000 retrieval_top_k: int 5 class Config: env_file .env settings Settings()4.2 文档入库切分、向量化、存储文档入库是RAG系统的第一步。我以Markdown格式的技术文档为例走一遍完整流程。第一步是读取文档。如果是PDF用pymupdf或者pdfplumber提取文本。如果是Markdown直接读取。如果是网页用trafilatura提取正文。提取的时候要注意保留结构信息比如标题层级、代码块、表格。第二步是切分。我的策略是两级切分先按标题切分成大块再按段落切分成小块。大块保留章节上下文小块用于检索。每个小块的大小控制在300到500字之间相邻块重叠50到100字。# scripts/ingest.py 的核心逻辑 def chunk_document(text, max_chunk_size500, overlap100): # 先按标题切分 sections split_by_headings(text) chunks [] for section in sections: # 再按段落切分 paragraphs section.content.split(\n\n) current_chunk for para in paragraphs: if len(current_chunk) len(para) max_chunk_size: chunks.append({ text: current_chunk, metadata: { section_title: section.title, source: section.source } }) # 保留重叠部分 current_chunk current_chunk[-overlap:] para else: current_chunk \n\n para if current_chunk: chunks.append({ text: current_chunk, metadata: { section_title: section.title, source: section.source } }) return chunks第三步是向量化。用embedding模型把每个chunk转成向量。这里要注意embedding模型和后面检索时用的模型必须一致否则向量空间不对齐检索结果会很差。我通常用BGE-M3或者text-embedding-3-small前者可以本地部署后者方便但需要API调用。第四步是存储。在PostgreSQL里建一个表包含chunk_id、text、metadata、embedding四个字段。embedding字段用pgvector的vector类型。建一个IVFFlat或者HNSW索引加速检索。CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE document_chunks ( id SERIAL PRIMARY KEY, text TEXT NOT NULL, metadata JSONB, embedding vector(1536) ); CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);注意入库的时候要记录文档的版本和入库时间。后面如果文档更新了你需要知道哪些chunk是旧的需要重新生成。我的做法是给每个chunk加一个doc_version字段检索的时候只查最新版本。4.3 检索与生成一次完整的请求处理用户提问进来之后系统需要经过检索、编排、生成三个主要步骤。我以“如何配置数据库连接池”这个问题为例走一遍完整流程。第一步查询理解。用户的原始问题可能很口语化直接拿去检索效果不好。我会先用一个小模型或者规则做查询改写把“如何配置数据库连接池”改写成“数据库连接池 配置 参数 设置”。同时提取关键词用于后面的混合检索。第二步混合检索。用向量检索从PgVector里找语义相似的chunk用关键词检索PostgreSQL的全文检索找精确匹配的chunk。然后做RRF融合取top 10。# app/retrieval/hybrid.py def hybrid_search(query, top_k10): # 向量检索 query_embedding get_embedding(query) vector_results db.execute( SELECT id, text, metadata, 1 - (embedding %s) as score FROM document_chunks ORDER BY embedding %s LIMIT %s , (query_embedding, query_embedding, top_k * 2)) # 关键词检索 keyword_results db.execute( SELECT id, text, metadata, ts_rank(to_tsvector(simple, text), plainto_tsquery(simple, %s)) as score FROM document_chunks WHERE to_tsvector(simple, text) plainto_tsquery(simple, %s) ORDER BY score DESC LIMIT %s , (query, query, top_k * 2)) # RRF融合 return rrf_fusion(vector_results, keyword_results, top_k)第三步重排序。取top 10之后用一个交叉编码器做精排。交叉编码器同时看query和chunk给出相关性分数比向量检索的精度高很多。我通常用BGE-reranker-base本地部署推理速度快。第四步构造prompt。把精排后的top 5 chunk拼成上下文加上系统提示词和用户问题构造最终的prompt。系统提示词要明确告诉模型只根据提供的上下文回答如果上下文没有相关信息就说不知道不要编造。SYSTEM_PROMPT 你是一个技术文档助手。请根据以下提供的文档片段回答用户问题。 规则 1. 只使用提供的文档片段中的信息回答 2. 如果文档片段中没有相关信息直接说根据现有文档无法回答该问题 3. 回答时引用具体的文档来源 4. 保持回答简洁准确 文档片段 {context} 第五步调用模型生成回答。用流式输出提升用户体验。同时记录token消耗和耗时用于后续的成本分析和性能优化。4.4 效果评估怎么知道你的系统好不好系统搭起来了怎么知道它好不好不能靠感觉要有量化的评估。我通常从三个维度来评估检索质量、生成质量、系统性能。检索质量用召回率和精确率来衡量。准备一批测试问题每个问题标注哪些chunk是相关的。然后看系统检索出来的top k里有多少是相关的。召回率低说明检索策略有问题精确率低说明排序有问题。生成质量用人工评估或者模型评估。人工评估最准但成本高。模型评估可以用GPT-4做裁判给生成的回答打分。我通常用两个指标忠实度回答是否基于检索到的内容没有编造和相关性回答是否切题。系统性能看延迟和成本。延迟包括检索延迟和生成延迟分别统计P50、P95、P99。成本看每次请求的token消耗和对应的费用。这些数据都要持续监控发现异常及时排查。# scripts/evaluate.py 的评估逻辑 def evaluate_retrieval(test_cases, top_k5): recall_scores [] precision_scores [] for case in test_cases: results hybrid_search(case.query, top_k) retrieved_ids [r.id for r in results] relevant_ids set(case.relevant_chunk_ids) hits len(set(retrieved_ids) relevant_ids) recall hits / len(relevant_ids) if relevant_ids else 0 precision hits / len(retrieved_ids) if retrieved_ids else 0 recall_scores.append(recall) precision_scores.append(precision) return { avg_recall: sum(recall_scores) / len(recall_scores), avg_precision: sum(precision_scores) / len(precision_scores) }实操心得评估集要覆盖各种情况简单问题、复杂问题、多跳问题、无法回答的问题。我见过很多项目只测简单问题上线后遇到复杂问题就崩了。另外评估要定期做每次修改检索策略或者换模型之后都要重新跑一遍确保没有退化。5. 常见问题与排查技巧实录5.1 检索不准从查询改写和切分策略入手检索不准是最常见的问题。用户问“怎么设置超时时间”检索出来的却是“超时错误处理”。这种情况通常是查询和文档的表述方式不一致导致的。我的排查思路是先看查询改写有没有问题。如果原始查询太短或者太口语化改写后的查询可能丢失了关键信息。可以尝试多种改写策略比如同义词扩展、关键词提取、生成多个查询变体然后合并结果。再看切分策略。如果chunk太大一个chunk里包含多个主题检索时匹配度就不高。如果chunk太小信息不完整生成时又不够用。我的经验是技术文档的chunk大小在300到500字比较合适问答对可以更小教程类可以更大。还有一个容易被忽略的点embedding模型的选择。不同模型对不同领域的文本表现差异很大。通用模型在技术文档上可能不如领域微调的模型。如果检索效果一直不理想可以考虑用领域数据微调embedding模型或者换一个在技术领域表现更好的模型。5.2 生成胡编用引用和校验来约束模型胡编是另一个高频问题。用户问了一个文档里没有的问题模型不直接说不知道而是根据训练数据编了一个答案。这在技术文档场景下特别危险因为用户可能真的照着错误答案去操作。我的解决方案是三层约束。第一层在prompt里明确要求“只根据提供的上下文回答没有相关信息就说不知道”。第二层要求模型在回答中引用具体的文档来源比如“根据《数据库配置指南》第3.2节”。第三层生成之后做一个校验检查回答中的关键信息是否能在检索到的chunk中找到。如果找不到就标记为“可能不准确”或者直接返回“无法回答”。def verify_answer(answer, retrieved_chunks): # 提取回答中的关键实体和数字 entities extract_entities(answer) numbers extract_numbers(answer) # 检查是否在检索内容中出现 context .join([c.text for c in retrieved_chunks]) missing [] for entity in entities numbers: if entity not in context: missing.append(entity) if missing: return False, f以下信息未在文档中找到: {missing} return True, 验证通过注意校验不要太严格否则会把正确的回答也拦掉。模型可能会用不同的表述方式或者做一些合理的推理。校验的目的是拦住明显的编造不是要求逐字匹配。5.3 延迟太高从并行化和缓存找空间延迟高是影响用户体验的直接因素。一个请求如果超过5秒用户就会觉得卡。超过10秒用户可能直接关掉页面。排查延迟先看瓶颈在哪。是检索慢还是生成慢检索慢通常是向量检索的索引没建好或者数据量太大。生成慢通常是模型本身的速度限制或者上下文太长。我的优化手段有几个。第一并行化。检索和查询改写可以并行做不用等改写完成再检索。多个检索策略也可以并行执行。第二缓存。相同或相似的查询检索结果可以缓存。embedding计算也可以缓存避免重复计算。第三流式输出。生成阶段用流式输出用户能很快看到第一个token感知延迟会低很多。第四模型降级。对延迟敏感的场景可以用小模型或者量化模型牺牲一点质量换速度。# 并行检索的示例 import asyncio async def parallel_retrieval(query): tasks [ vector_search(query), keyword_search(query), # 可以加更多检索策略 ] results await asyncio.gather(*tasks) return merge_results(results)5.4 成本失控监控、限流、缓存三管齐下AI应用的成本主要来自模型调用。如果不加控制一个爆款功能可能一夜之间烧掉几个月的预算。我见过最夸张的案例一个没有限流的AI功能被爬虫刷了一天消耗了上万美元。成本控制的第一道防线是监控。每次模型调用都要记录token消耗和费用按小时、按天、按用户维度统计。设置告警阈值比如日消耗超过预算的80%就发通知。第二道防线是限流。按用户、按IP、按接口做限流。免费用户每天限制调用次数付费用户根据套餐限制。限流策略要灵活可以动态调整。第三道防线是缓存。相同的问题不要重复调用模型。缓存可以设在多个层级精确匹配缓存、语义相似缓存、检索结果缓存。语义相似缓存用embedding做相似度匹配相似度超过阈值就复用之前的回答。# 语义缓存的简化实现 class SemanticCache: def __init__(self, threshold0.95): self.cache [] # 存储 (embedding, response) self.threshold threshold def get(self, query_embedding): for cached_embedding, response in self.cache: similarity cosine_similarity(query_embedding, cached_embedding) if similarity self.threshold: return response return None def set(self, query_embedding, response): self.cache.append((query_embedding, response))5.5 常见问题速查表问题现象可能原因排查方向解决方案检索结果不相关查询改写丢失信息对比原始查询和改写后查询调整改写策略保留关键词检索结果不相关chunk切分不合理检查chunk大小和边界调整切分参数增加重叠生成内容胡编prompt约束不够检查系统提示词加强约束增加引用要求生成内容胡编检索内容质量差人工检查检索结果优化检索策略增加重排序延迟高检索慢分析各阶段耗时建索引并行化缓存延迟高生成慢检查上下文长度减少上下文流式输出成本高重复调用多分析调用日志加缓存限流成本高上下文太长统计token消耗压缩上下文精简prompt多轮对话混乱历史管理不当检查对话状态引入摘要和长期记忆输出格式错误模型不遵循schema检查prompt示例增加校验和修复层实操心得排查问题的时候一定要有完整的日志。我建议每次请求都记录一个trace_id把检索结果、prompt、模型输出、耗时、token消耗全部关联起来。出问题的时候用trace_id一查整个链路清清楚楚。没有日志的排查就是盲人摸象。6. 我踩过的那些坑和后来想明白的事做AI工程这几年踩过的坑太多了挑几个印象深刻的说说。第一个坑是过度依赖框架。刚开始做RAG的时候我用了LangChain觉得它什么都有省事。结果上线后发现检索效果不好想调一下切分策略发现LangChain的切分器封装得太深改起来很麻烦。想加一个自定义的检索策略发现要继承一堆类理解成本很高。后来我干脆把LangChain去掉自己写了一套轻量的编排逻辑代码量不大但每个环节都清清楚楚想改哪里改哪里。我的体会是框架适合快速原型但生产系统还是要把核心逻辑掌握在自己手里。第二个坑是忽略数据质量。有段时间我一直在调模型和prompt但效果就是上不去。后来花了一天时间人工检查检索回来的内容发现知识库里有大量过时的、重复的、格式混乱的文档。把这些数据清理之后同样的模型和prompt效果提升了一大截。这件事让我明白AI工程里数据质量比模型选择重要得多。垃圾进垃圾出这个道理在AI时代依然成立。第三个坑是不做评估。早期做项目改完代码凭感觉觉得“好像好了一点”就上线了。结果有时候改了一个地方另一个地方退化了自己还不知道。后来我建了一个小规模的评估集每次改动都跑一遍用数据说话。虽然建评估集要花时间但长期来看节省的时间更多因为你能快速知道一个改动是正向还是负向的。第四个坑是低估了运维的复杂度。模型API会限流、会超时、会涨价、会下线。向量数据库会爆内存、会索引失效。这些在生产环境里都是真实会发生的问题。我的建议是从第一天起就要考虑容错和降级。主模型挂了有没有备用检索服务挂了能不能降级到关键词检索这些预案要在设计阶段就想好不要等出了问题再临时抱佛脚。最后分享一个我觉得很有用的习惯每次做完一个项目把关键的决策、踩过的坑、有效的解决方案整理成文档。不是为了给别人看是为了给自己看。下一个项目遇到类似问题的时候翻出来看看能少走很多弯路。AI工程这个领域变化太快但很多底层的东西是不变的。把不变的东西沉淀下来才能跟上变化的速度。这个“ai-engineering-from-scratch”的项目后续还可以往几个方向扩展。比如加入多模态能力支持图片和表格的检索。比如加入Agent能力让系统能调用外部工具完成更复杂的任务。比如加入微调环节用领域数据提升模型在特定任务上的表现。每一个方向都值得单独写一篇但核心思路是一样的从数据流出发把每个环节做扎实用评估驱动优化。 从零搭建AI工程体系：架构设计、核心模块与避坑指南
1. 从零搭建AI工程体系为什么我劝你别一上来就调包“ai-engineering-from-scratch”这个标题第一次看到的时候我愣了一下。不是因为陌生恰恰相反是因为它戳中了我这几年带团队、面试、做项目咨询时反复遇到的一个痛点太多人把“会调API”当成了“会做AI工程”。我见过不少简历上写着“熟悉大模型应用开发”的同学聊下来发现他们的全部经验就是pip install openai然后写个while循环拼prompt。一旦线上QPS上来、上下文变长、多轮对话状态需要管理、模型输出格式不稳定、成本失控立刻就懵了。这不是他们的错因为市面上绝大多数教程都在教“怎么用”很少有人系统性地讲“怎么从零把一套AI工程体系搭起来”。所以当我看到“ai-engineering-from-scratch”这个项目标题时我脑子里立刻浮现出一个清晰的画像这是一个面向有一定编程基础、但希望真正理解AI工程全貌的开发者从最底层开始一步步构建出可运行、可扩展、可维护的AI应用系统的实践项目。它要解决的核心问题不是“怎么调用某个模型”而是“当你要把AI能力变成一个真正的产品功能时中间到底需要哪些环节每个环节的关键决策是什么”。这篇文章我就围绕这个项目标题把我自己从零搭建AI工程体系的经验、踩过的坑、以及那些文档里不会写的细节完整地拆解一遍。适合谁看如果你已经会写Python用过至少一个AI模型的API但总觉得自己的项目“能跑但不敢上线”那这篇内容就是为你准备的。如果你是完全的新手也没关系我会把每个环节的“为什么”讲清楚让你知道每一步在解决什么问题。2. 整体架构设计先想清楚数据怎么流再动手写代码2.1 为什么我不建议从模型选型开始很多人做AI项目第一反应是“我该用哪个模型”。GPT-4还是Claude开源模型能不能打要不要微调这些问题当然重要但它们不是起点。真正的起点是你的数据怎么流。我举个具体的例子。假设你要做一个“智能客服助手”能根据用户问题从知识库检索答案并生成回复。如果你一上来就纠结用哪个模型你可能会花三天时间对比各种模型的benchmark最后选了一个“最强”的。但真正上线后你会发现瓶颈根本不在模型而在检索环节——知识库的切分策略、向量化的质量、召回率的高低这些才是决定用户体验的关键。模型再强检索回来的内容不对生成的结果就是胡编乱造。所以“ai-engineering-from-scratch”这个项目我建议的架构设计顺序是先定义数据流再确定组件边界最后才做技术选型。数据流是什么就是用户输入进来之后经过哪些处理步骤最终输出什么结果。每个步骤需要什么数据、产生什么数据、数据量级大概多少、延迟要求是多少。把这些想清楚技术选型就是水到渠成的事。2.2 一个可落地的分层架构参考基于我自己的实践经验一个从零搭建的AI工程体系可以分成四层。这个分层不是理论模型是我在实际项目中反复调整后觉得最顺手的结构。接入层负责接收用户请求做初步的参数校验和限流。这一层不需要太复杂一个FastAPI或者Flask应用就够了。关键是做好请求日志把每次请求的输入、输出、耗时、token消耗都记下来。这些日志后面做成本分析和效果优化时就是金矿。编排层这是整个系统的核心。它负责决定“这次请求要走哪条路径”。比如用户问的是事实性问题走检索增强生成RAG路径用户问的是需要计算的问题走工具调用路径用户只是闲聊走直接生成路径。编排层还需要管理多轮对话的状态决定哪些历史信息要带入下一轮。这一层我通常用一个状态机或者简单的规则引擎来实现不一定非要上LangChain那种重框架。能力层这一层才是模型、检索器、工具函数等具体能力的实现。模型调用要做统一的封装包括重试、降级、超时控制。检索器要支持多种索引方式比如向量检索、关键词检索、混合检索。工具函数要定义清晰的输入输出schema方便编排层调用。数据层包括向量数据库、关系型数据库、缓存、对象存储等。这一层的关键是数据的一致性和可追溯性。比如每次检索用到的知识库版本要记录每次模型调用的参数要留存方便后面做A/B测试和问题排查。注意这个分层不是绝对的。小项目可以合并大项目可以再拆。但核心思想是每一层只做自己该做的事层与层之间通过清晰的接口通信。这样后面任何一层要换实现都不会影响其他层。2.3 技术选型背后的取舍逻辑有了架构分层技术选型就有章可循了。我拿几个关键决策点来说说我的取舍逻辑。模型调用方式直接用官方SDK还是自己封装HTTP请求我的建议是初期用官方SDK快速验证但一定要在SDK外面再包一层自己的接口。为什么因为你需要统一处理重试、超时、降级、日志。如果直接散落在业务代码里调SDK后面想换模型供应商或者加个缓存就得改遍所有地方。我自己的做法是定义一个ModelClient抽象类所有模型调用都通过它具体实现可以是OpenAI、Anthropic或者本地模型。向量数据库选型Chroma、Qdrant、Milvus、PgVector选哪个我的经验是数据量在百万级以下、团队没有专职运维的情况下PgVector是最省心的选择。它直接跑在PostgreSQL里你不需要额外维护一个数据库服务备份、监控、权限管理都能复用现有的PostgreSQL工具链。数据量上亿了再考虑Milvus或者Qdrant。Chroma适合本地开发和小规模原型但生产环境我不太推荐主要是持久化和并发方面需要额外考虑。编排框架LangChain、LlamaIndex还是自己写我的观点可能有点反主流如果你是要做一个长期维护的产品我建议自己写编排逻辑。LangChain的抽象层次太多出了问题很难调试而且版本更新频繁升级一次可能改一堆代码。自己写编排逻辑代码量其实不大但可控性完全不一样。当然如果你只是做原型验证LangChain能帮你省不少时间。部署方式容器化是必须的。Docker Compose适合单机部署Kubernetes适合多服务、需要弹性伸缩的场景。我的建议是即使初期只用一台服务器也要用Docker Compose把各个服务拆开。这样后面要迁移到K8s或者加服务成本会低很多。3. 核心模块拆解每个环节的坑我都替你踩过了3.1 模型调用封装别让重试逻辑散落各处模型调用看起来简单不就是发个HTTP请求吗但实际生产环境中你需要处理的情况远比想象中多。网络抖动、API限流、模型返回格式错误、超时、余额不足每一种情况都需要不同的处理策略。我自己的ModelClient封装通常包含这几个关键设计。第一统一的重试策略。对于网络错误和限流错误用指数退避重试对于参数错误和余额不足直接失败不重试。第二超时控制。每个请求设置一个合理的超时时间比如30秒超时后走降级逻辑。第三降级策略。主模型不可用时自动切换到备用模型或者返回一个兜底回复。第四日志记录。每次调用的输入、输出、耗时、token消耗、是否重试、最终状态全部结构化记录。这里有个细节很多人会忽略token消耗的计算。不同模型的token计算方式不一样而且流式输出和非流式输出的token统计也有差异。我建议在封装层统一做token估算用tiktoken这样的库虽然不完全准确但足够做成本监控了。精确的token消耗可以从API返回的usage字段获取但流式输出时很多API不返回usage这时候就需要自己估算。# 一个简化的ModelClient封装示例 class ModelClient: def __init__(self, primary_model, fallback_modelNone): self.primary primary_model self.fallback fallback_model self.max_retries 3 def call(self, messages, **kwargs): for attempt in range(self.max_retries): try: response self._call_model(self.primary, messages, **kwargs) self._log_success(response) return response except RateLimitError: time.sleep(2 ** attempt) except (NetworkError, TimeoutError): time.sleep(1) except ModelError as e: self._log_error(e) break if self.fallback: return self._call_model(self.fallback, messages, **kwargs) raise ModelUnavailableError(All models failed)实操心得重试次数不要设太多3次足够了。我见过有人设10次重试结果一个请求卡了半分钟用户体验极差。另外重试的时候要考虑幂等性如果模型调用有副作用比如写数据库重试前要确认是否已经执行过。3.2 检索增强生成切分策略决定成败RAG是现在AI工程里最常用的模式但也是最容易做砸的环节。我见过太多项目模型选的是最好的prompt写得也很用心但检索回来的内容驴唇不对马嘴生成的结果自然好不了。RAG的核心在于检索质量而检索质量的第一道关卡是文档切分。切分策略没有万能公式但有几个原则可以参考。第一按语义切分不要按固定字数切分。一个完整的段落、一个章节、一个问答对都是自然的语义单元。按固定字数切分很容易把一句话切成两半检索的时候两边都匹配不上。第二切分后的块要有适当的重叠。比如每个块500字相邻块之间重叠100字这样跨块的信息不会丢失。第三每个块要保留足够的元数据。来源文档、章节标题、页码、时间戳这些信息在生成阶段可以用来做引用和过滤。向量化模型的选择也很关键。我的经验是中文场景下BGE系列或者M3E系列的效果通常比OpenAI的text-embedding-ada-002好。而且这些模型可以本地部署没有API调用成本数据也不用出本地。当然如果你的数据量特别大本地部署的推理速度可能是个瓶颈这时候可以考虑用GPU加速或者用API。检索策略上我强烈建议用混合检索向量检索加关键词检索然后做融合排序。纯向量检索对语义匹配好但对精确匹配比如产品型号、人名容易漏。关键词检索比如BM25对精确匹配好但对语义变体不行。两者结合召回率能提升不少。融合排序可以用RRFReciprocal Rank Fusion简单有效不需要训练。# RRF融合排序的简化实现 def rrf_fusion(vector_results, keyword_results, k60): scores {} for rank, doc_id in enumerate(vector_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) for rank, doc_id in enumerate(keyword_results): scores[doc_id] scores.get(doc_id, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)注意检索回来的内容不要一股脑全塞给模型。上下文长度是有限的而且塞太多无关内容反而会干扰模型。我的做法是先做一轮粗排取top 20然后用一个小的交叉编码器做精排取top 5送给模型。交叉编码器可以用BGE-reranker效果很明显。3.3 对话状态管理多轮对话不是简单拼历史多轮对话是AI应用里最复杂的部分之一。很多人以为多轮对话就是把历史消息拼在一起发给模型但实际做起来会发现一堆问题历史太长超出上下文限制、用户切换话题后旧信息干扰、指代消解“它”、“那个”指什么需要额外处理。我的做法是把对话状态分成几个部分来管理。短期记忆最近几轮对话的原始消息直接拼在prompt里。长期记忆从历史对话中提取的关键信息比如用户的名字、偏好、之前提到的事实用结构化的方式存储需要时再注入prompt。会话摘要当对话轮次超过一定数量时把早期的对话压缩成摘要替代原始消息节省上下文空间。指代消解是个容易被忽略的问题。用户说“帮我查一下它的价格”这个“它”指的是上一轮提到的商品。如果直接把这句话发给模型模型可能不知道“它”是什么。我的做法是在编排层加一个轻量的指代消解步骤用规则或者小模型把指代词替换成具体实体再发给主模型。这个步骤不需要太精确但能显著提升多轮对话的准确率。会话状态的存储我建议用Redis或者内存数据库设置合理的过期时间。不要存在关系型数据库里因为读写太频繁而且状态数据通常不需要长期持久化。如果需要做对话分析可以异步把状态快照写到数据仓库。3.4 输出解析与校验别相信模型会乖乖返回JSON如果你让模型返回JSON格式十次里可能有八次是合法的但剩下两次会给你各种惊喜多一个逗号、少一个引号、用单引号、嵌套结构不对、甚至直接返回一段解释文字。在生产环境里这种不确定性是致命的。我的做法是三层防护。第一层在prompt里明确要求返回JSON并给出schema示例。第二层用JSON解析器尝试解析如果失败尝试修复常见错误比如去掉尾随逗号、替换单引号。第三层如果还是失败用一个小的修复模型或者规则引擎把模型的输出重新格式化成合法JSON。如果三层都失败走降级逻辑返回一个默认值或者错误提示。对于关键字段还要做业务校验。比如模型返回了一个日期你要检查这个日期是否合法、是否在合理范围内。模型返回了一个金额你要检查是否为正数、是否超出限额。这些校验看起来琐碎但能避免很多线上事故。# 输出解析的三层防护示例 def parse_model_output(raw_output, schema): # 第一层直接解析 try: data json.loads(raw_output) return validate_schema(data, schema) except json.JSONDecodeError: pass # 第二层修复常见错误 try: fixed fix_common_json_errors(raw_output) data json.loads(fixed) return validate_schema(data, schema) except (json.JSONDecodeError, ValidationError): pass # 第三层用规则提取关键字段 try: data extract_fields_by_regex(raw_output, schema) return validate_schema(data, schema) except ExtractionError: raise OutputParseError(Failed to parse model output)实操心得在prompt里给schema示例的时候尽量用具体的值不要用占位符。比如写{name: 张三, age: 30}比写{name: string, age: number}效果好得多。模型对具体示例的遵循度更高。4. 实操全流程从零到一搭建一个可运行的AI工程原型4.1 环境准备与项目初始化说了这么多设计层面的东西现在来点实际的。我带你走一遍从零搭建一个AI工程原型的完整流程。这个原型是一个“技术文档问答助手”用户可以提问系统从技术文档中检索相关内容并生成回答。首先是环境准备。Python 3.10以上推荐用uv或者poetry管理依赖比pip快很多。核心依赖包括FastAPI做Web框架OpenAI或者Anthropic的SDK做模型调用PgVector做向量存储tiktoken做token计算pydantic做数据校验。# 用uv初始化项目 uv init ai-engineering-from-scratch cd ai-engineering-from-scratch uv add fastapi uvicorn openai psycopg2-binary pgvector tiktoken pydantic python-dotenv项目结构我建议这样组织ai-engineering-from-scratch/ ├── app/ │ ├── main.py # FastAPI入口 │ ├── config.py # 配置管理 │ ├── models/ # 数据模型 │ ├── clients/ # 模型客户端封装 │ ├── retrieval/ # 检索模块 │ ├── orchestration/ # 编排逻辑 │ └── utils/ # 工具函数 ├── scripts/ │ ├── ingest.py # 文档入库脚本 │ └── evaluate.py # 效果评估脚本 ├── tests/ ├── docker-compose.yml └── .env配置管理用pydantic的BaseSettings把模型API密钥、数据库连接串、各种超时参数都放在环境变量里。不要硬编码在代码里也不要提交到git。# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str database_url: str model_name: str gpt-4o-mini embedding_model: str text-embedding-3-small max_context_tokens: int 4000 retrieval_top_k: int 5 class Config: env_file .env settings Settings()4.2 文档入库切分、向量化、存储文档入库是RAG系统的第一步。我以Markdown格式的技术文档为例走一遍完整流程。第一步是读取文档。如果是PDF用pymupdf或者pdfplumber提取文本。如果是Markdown直接读取。如果是网页用trafilatura提取正文。提取的时候要注意保留结构信息比如标题层级、代码块、表格。第二步是切分。我的策略是两级切分先按标题切分成大块再按段落切分成小块。大块保留章节上下文小块用于检索。每个小块的大小控制在300到500字之间相邻块重叠50到100字。# scripts/ingest.py 的核心逻辑 def chunk_document(text, max_chunk_size500, overlap100): # 先按标题切分 sections split_by_headings(text) chunks [] for section in sections: # 再按段落切分 paragraphs section.content.split(\n\n) current_chunk for para in paragraphs: if len(current_chunk) len(para) max_chunk_size: chunks.append({ text: current_chunk, metadata: { section_title: section.title, source: section.source } }) # 保留重叠部分 current_chunk current_chunk[-overlap:] para else: current_chunk \n\n para if current_chunk: chunks.append({ text: current_chunk, metadata: { section_title: section.title, source: section.source } }) return chunks第三步是向量化。用embedding模型把每个chunk转成向量。这里要注意embedding模型和后面检索时用的模型必须一致否则向量空间不对齐检索结果会很差。我通常用BGE-M3或者text-embedding-3-small前者可以本地部署后者方便但需要API调用。第四步是存储。在PostgreSQL里建一个表包含chunk_id、text、metadata、embedding四个字段。embedding字段用pgvector的vector类型。建一个IVFFlat或者HNSW索引加速检索。CREATE EXTENSION IF NOT EXISTS vector; CREATE TABLE document_chunks ( id SERIAL PRIMARY KEY, text TEXT NOT NULL, metadata JSONB, embedding vector(1536) ); CREATE INDEX ON document_chunks USING hnsw (embedding vector_cosine_ops);注意入库的时候要记录文档的版本和入库时间。后面如果文档更新了你需要知道哪些chunk是旧的需要重新生成。我的做法是给每个chunk加一个doc_version字段检索的时候只查最新版本。4.3 检索与生成一次完整的请求处理用户提问进来之后系统需要经过检索、编排、生成三个主要步骤。我以“如何配置数据库连接池”这个问题为例走一遍完整流程。第一步查询理解。用户的原始问题可能很口语化直接拿去检索效果不好。我会先用一个小模型或者规则做查询改写把“如何配置数据库连接池”改写成“数据库连接池 配置 参数 设置”。同时提取关键词用于后面的混合检索。第二步混合检索。用向量检索从PgVector里找语义相似的chunk用关键词检索PostgreSQL的全文检索找精确匹配的chunk。然后做RRF融合取top 10。# app/retrieval/hybrid.py def hybrid_search(query, top_k10): # 向量检索 query_embedding get_embedding(query) vector_results db.execute( SELECT id, text, metadata, 1 - (embedding %s) as score FROM document_chunks ORDER BY embedding %s LIMIT %s , (query_embedding, query_embedding, top_k * 2)) # 关键词检索 keyword_results db.execute( SELECT id, text, metadata, ts_rank(to_tsvector(simple, text), plainto_tsquery(simple, %s)) as score FROM document_chunks WHERE to_tsvector(simple, text) plainto_tsquery(simple, %s) ORDER BY score DESC LIMIT %s , (query, query, top_k * 2)) # RRF融合 return rrf_fusion(vector_results, keyword_results, top_k)第三步重排序。取top 10之后用一个交叉编码器做精排。交叉编码器同时看query和chunk给出相关性分数比向量检索的精度高很多。我通常用BGE-reranker-base本地部署推理速度快。第四步构造prompt。把精排后的top 5 chunk拼成上下文加上系统提示词和用户问题构造最终的prompt。系统提示词要明确告诉模型只根据提供的上下文回答如果上下文没有相关信息就说不知道不要编造。SYSTEM_PROMPT 你是一个技术文档助手。请根据以下提供的文档片段回答用户问题。 规则 1. 只使用提供的文档片段中的信息回答 2. 如果文档片段中没有相关信息直接说根据现有文档无法回答该问题 3. 回答时引用具体的文档来源 4. 保持回答简洁准确 文档片段 {context} 第五步调用模型生成回答。用流式输出提升用户体验。同时记录token消耗和耗时用于后续的成本分析和性能优化。4.4 效果评估怎么知道你的系统好不好系统搭起来了怎么知道它好不好不能靠感觉要有量化的评估。我通常从三个维度来评估检索质量、生成质量、系统性能。检索质量用召回率和精确率来衡量。准备一批测试问题每个问题标注哪些chunk是相关的。然后看系统检索出来的top k里有多少是相关的。召回率低说明检索策略有问题精确率低说明排序有问题。生成质量用人工评估或者模型评估。人工评估最准但成本高。模型评估可以用GPT-4做裁判给生成的回答打分。我通常用两个指标忠实度回答是否基于检索到的内容没有编造和相关性回答是否切题。系统性能看延迟和成本。延迟包括检索延迟和生成延迟分别统计P50、P95、P99。成本看每次请求的token消耗和对应的费用。这些数据都要持续监控发现异常及时排查。# scripts/evaluate.py 的评估逻辑 def evaluate_retrieval(test_cases, top_k5): recall_scores [] precision_scores [] for case in test_cases: results hybrid_search(case.query, top_k) retrieved_ids [r.id for r in results] relevant_ids set(case.relevant_chunk_ids) hits len(set(retrieved_ids) relevant_ids) recall hits / len(relevant_ids) if relevant_ids else 0 precision hits / len(retrieved_ids) if retrieved_ids else 0 recall_scores.append(recall) precision_scores.append(precision) return { avg_recall: sum(recall_scores) / len(recall_scores), avg_precision: sum(precision_scores) / len(precision_scores) }实操心得评估集要覆盖各种情况简单问题、复杂问题、多跳问题、无法回答的问题。我见过很多项目只测简单问题上线后遇到复杂问题就崩了。另外评估要定期做每次修改检索策略或者换模型之后都要重新跑一遍确保没有退化。5. 常见问题与排查技巧实录5.1 检索不准从查询改写和切分策略入手检索不准是最常见的问题。用户问“怎么设置超时时间”检索出来的却是“超时错误处理”。这种情况通常是查询和文档的表述方式不一致导致的。我的排查思路是先看查询改写有没有问题。如果原始查询太短或者太口语化改写后的查询可能丢失了关键信息。可以尝试多种改写策略比如同义词扩展、关键词提取、生成多个查询变体然后合并结果。再看切分策略。如果chunk太大一个chunk里包含多个主题检索时匹配度就不高。如果chunk太小信息不完整生成时又不够用。我的经验是技术文档的chunk大小在300到500字比较合适问答对可以更小教程类可以更大。还有一个容易被忽略的点embedding模型的选择。不同模型对不同领域的文本表现差异很大。通用模型在技术文档上可能不如领域微调的模型。如果检索效果一直不理想可以考虑用领域数据微调embedding模型或者换一个在技术领域表现更好的模型。5.2 生成胡编用引用和校验来约束模型胡编是另一个高频问题。用户问了一个文档里没有的问题模型不直接说不知道而是根据训练数据编了一个答案。这在技术文档场景下特别危险因为用户可能真的照着错误答案去操作。我的解决方案是三层约束。第一层在prompt里明确要求“只根据提供的上下文回答没有相关信息就说不知道”。第二层要求模型在回答中引用具体的文档来源比如“根据《数据库配置指南》第3.2节”。第三层生成之后做一个校验检查回答中的关键信息是否能在检索到的chunk中找到。如果找不到就标记为“可能不准确”或者直接返回“无法回答”。def verify_answer(answer, retrieved_chunks): # 提取回答中的关键实体和数字 entities extract_entities(answer) numbers extract_numbers(answer) # 检查是否在检索内容中出现 context .join([c.text for c in retrieved_chunks]) missing [] for entity in entities numbers: if entity not in context: missing.append(entity) if missing: return False, f以下信息未在文档中找到: {missing} return True, 验证通过注意校验不要太严格否则会把正确的回答也拦掉。模型可能会用不同的表述方式或者做一些合理的推理。校验的目的是拦住明显的编造不是要求逐字匹配。5.3 延迟太高从并行化和缓存找空间延迟高是影响用户体验的直接因素。一个请求如果超过5秒用户就会觉得卡。超过10秒用户可能直接关掉页面。排查延迟先看瓶颈在哪。是检索慢还是生成慢检索慢通常是向量检索的索引没建好或者数据量太大。生成慢通常是模型本身的速度限制或者上下文太长。我的优化手段有几个。第一并行化。检索和查询改写可以并行做不用等改写完成再检索。多个检索策略也可以并行执行。第二缓存。相同或相似的查询检索结果可以缓存。embedding计算也可以缓存避免重复计算。第三流式输出。生成阶段用流式输出用户能很快看到第一个token感知延迟会低很多。第四模型降级。对延迟敏感的场景可以用小模型或者量化模型牺牲一点质量换速度。# 并行检索的示例 import asyncio async def parallel_retrieval(query): tasks [ vector_search(query), keyword_search(query), # 可以加更多检索策略 ] results await asyncio.gather(*tasks) return merge_results(results)5.4 成本失控监控、限流、缓存三管齐下AI应用的成本主要来自模型调用。如果不加控制一个爆款功能可能一夜之间烧掉几个月的预算。我见过最夸张的案例一个没有限流的AI功能被爬虫刷了一天消耗了上万美元。成本控制的第一道防线是监控。每次模型调用都要记录token消耗和费用按小时、按天、按用户维度统计。设置告警阈值比如日消耗超过预算的80%就发通知。第二道防线是限流。按用户、按IP、按接口做限流。免费用户每天限制调用次数付费用户根据套餐限制。限流策略要灵活可以动态调整。第三道防线是缓存。相同的问题不要重复调用模型。缓存可以设在多个层级精确匹配缓存、语义相似缓存、检索结果缓存。语义相似缓存用embedding做相似度匹配相似度超过阈值就复用之前的回答。# 语义缓存的简化实现 class SemanticCache: def __init__(self, threshold0.95): self.cache [] # 存储 (embedding, response) self.threshold threshold def get(self, query_embedding): for cached_embedding, response in self.cache: similarity cosine_similarity(query_embedding, cached_embedding) if similarity self.threshold: return response return None def set(self, query_embedding, response): self.cache.append((query_embedding, response))5.5 常见问题速查表问题现象可能原因排查方向解决方案检索结果不相关查询改写丢失信息对比原始查询和改写后查询调整改写策略保留关键词检索结果不相关chunk切分不合理检查chunk大小和边界调整切分参数增加重叠生成内容胡编prompt约束不够检查系统提示词加强约束增加引用要求生成内容胡编检索内容质量差人工检查检索结果优化检索策略增加重排序延迟高检索慢分析各阶段耗时建索引并行化缓存延迟高生成慢检查上下文长度减少上下文流式输出成本高重复调用多分析调用日志加缓存限流成本高上下文太长统计token消耗压缩上下文精简prompt多轮对话混乱历史管理不当检查对话状态引入摘要和长期记忆输出格式错误模型不遵循schema检查prompt示例增加校验和修复层实操心得排查问题的时候一定要有完整的日志。我建议每次请求都记录一个trace_id把检索结果、prompt、模型输出、耗时、token消耗全部关联起来。出问题的时候用trace_id一查整个链路清清楚楚。没有日志的排查就是盲人摸象。6. 我踩过的那些坑和后来想明白的事做AI工程这几年踩过的坑太多了挑几个印象深刻的说说。第一个坑是过度依赖框架。刚开始做RAG的时候我用了LangChain觉得它什么都有省事。结果上线后发现检索效果不好想调一下切分策略发现LangChain的切分器封装得太深改起来很麻烦。想加一个自定义的检索策略发现要继承一堆类理解成本很高。后来我干脆把LangChain去掉自己写了一套轻量的编排逻辑代码量不大但每个环节都清清楚楚想改哪里改哪里。我的体会是框架适合快速原型但生产系统还是要把核心逻辑掌握在自己手里。第二个坑是忽略数据质量。有段时间我一直在调模型和prompt但效果就是上不去。后来花了一天时间人工检查检索回来的内容发现知识库里有大量过时的、重复的、格式混乱的文档。把这些数据清理之后同样的模型和prompt效果提升了一大截。这件事让我明白AI工程里数据质量比模型选择重要得多。垃圾进垃圾出这个道理在AI时代依然成立。第三个坑是不做评估。早期做项目改完代码凭感觉觉得“好像好了一点”就上线了。结果有时候改了一个地方另一个地方退化了自己还不知道。后来我建了一个小规模的评估集每次改动都跑一遍用数据说话。虽然建评估集要花时间但长期来看节省的时间更多因为你能快速知道一个改动是正向还是负向的。第四个坑是低估了运维的复杂度。模型API会限流、会超时、会涨价、会下线。向量数据库会爆内存、会索引失效。这些在生产环境里都是真实会发生的问题。我的建议是从第一天起就要考虑容错和降级。主模型挂了有没有备用检索服务挂了能不能降级到关键词检索这些预案要在设计阶段就想好不要等出了问题再临时抱佛脚。最后分享一个我觉得很有用的习惯每次做完一个项目把关键的决策、踩过的坑、有效的解决方案整理成文档。不是为了给别人看是为了给自己看。下一个项目遇到类似问题的时候翻出来看看能少走很多弯路。AI工程这个领域变化太快但很多底层的东西是不变的。把不变的东西沉淀下来才能跟上变化的速度。这个“ai-engineering-from-scratch”的项目后续还可以往几个方向扩展。比如加入多模态能力支持图片和表格的检索。比如加入Agent能力让系统能调用外部工具完成更复杂的任务。比如加入微调环节用领域数据提升模型在特定任务上的表现。每一个方向都值得单独写一篇但核心思路是一样的从数据流出发把每个环节做扎实用评估驱动优化。