尧图网站建设 
1. 从零搭建AI工程体系为什么我劝你别再东拼西凑这两年AI应用开发的门槛肉眼可见地降低了随便调个API就能跑出一个能对话的Demo但真正要把一个AI功能做成稳定、可维护、能扛住真实用户流量的工程系统绝大多数人卡在了同一个地方知识全是碎片。今天看了一篇讲RAG的帖子明天刷到一个讲向量数据库的视频后天又被人安利某个Agent框架结果真到自己动手的时候发现连一个完整的请求链路都串不起来。ai-engineering-from-scratch这个方向说白了就是解决这个问题的。它不是某一个具体的开源项目而是一套从底层原理到工程落地的完整学习与实践路径。核心目标是让你具备独立设计、开发、部署、运维一个AI应用系统的能力而不是停留在“会调API”的层面。这套东西适合谁我认为有三类人最该认真对待一是后端或全栈工程师想转型做AI应用开发二是有一定Python基础但没系统做过AI工程的数据分析人员三是技术负责人需要评估AI项目的可行性和技术选型。我自己的经历比较典型。两年前我开始接触大模型应用开发那时候市面上还没有现在这么多成熟的框架踩的坑一个接一个。最惨的一次是做一个文档问答系统本地测试效果很好上线之后并发一上来响应时间直接从2秒飙到30秒用户投诉不断。排查了两天才发现是向量检索那块没有做索引优化每次查询都在做全量暴力搜索。这件事让我意识到AI工程和传统后端工程虽然有交集但有很多独特的坑必须系统性地去理解和掌握。这篇文章我会把ai-engineering-from-scratch涉及的核心环节全部拆开讲从整体架构设计思路到每个模块的具体实现细节再到实操过程中会遇到的问题和排查方法。内容会比较长但如果你真的想把这套东西搞明白建议耐心看完。我会尽量用大白话解释每个技术选择背后的原因让你不仅知道怎么做更知道为什么这么做。2. 整体架构设计与技术选型思路2.1 为什么分层架构是AI工程的起点很多人做AI应用上来就开始写Prompt、调模型代码全堆在一个文件里。这种写法做Demo没问题但一旦需求变化或者要加新功能整个代码就会变成一团乱麻。我在早期项目里就吃过这个亏一个客服机器人项目最开始只有问答功能后来要加意图识别、多轮对话管理、知识库检索代码从200行膨胀到3000多行改一个地方要动全身。从零搭建AI工程体系第一件事就是建立清晰的分层架构。我推荐的分层方式是四层接入层、编排层、能力层、基础设施层。接入层负责处理用户请求的接收和响应的返回包括API网关、鉴权、限流这些传统后端的东西。编排层是整个系统的核心负责协调各个AI能力模块的调用顺序和逻辑比如先做意图识别再决定是走知识库检索还是直接调用模型生成。能力层是具体的AI功能实现包括大模型调用、向量检索、文本嵌入、重排序等等。基础设施层则是支撑上面所有功能的基础组件包括向量数据库、缓存、消息队列、监控系统。这样分层的好处是每一层可以独立演进。比如你想把底层的大模型从A换成B只需要改能力层的模型调用模块编排层和接入层完全不用动。再比如你想给系统加一个缓存机制也只需要在基础设施层加一个Redis然后在编排层里调用就行。注意分层不是目的解耦才是。不要为了分层而分层如果你的项目确实很简单一个单体应用就够了过度设计反而增加维护成本。2.2 模型选型的核心考量维度模型选型是AI工程里最容易被忽视但又极其关键的决策。我见过太多团队在这个问题上要么过于随意要么过于纠结。过于随意的是直接用最贵的模型觉得贵的就是好的过于纠结的是花几周时间做各种评测结果项目进度严重滞后。我的经验是模型选型要从四个维度来考量能力、成本、延迟、可控性。能力指的是模型在你具体任务上的表现这个必须用你自己的数据来测不能只看公开榜单。成本包括API调用费用和自部署的硬件成本要算清楚每千次请求的实际花费。延迟直接影响用户体验特别是需要实时响应的场景比如对话系统。可控性指的是你能不能对模型的行为做精细控制比如是否支持Function Calling、是否支持结构化输出、是否支持微调。具体怎么选我一般会准备一个包含50到100条真实场景数据的测试集然后拿几个候选模型跑一遍从准确率、响应时间、费用三个维度打分。下面这个表格是我最近一个项目里的实际评测结果供参考模型任务准确率平均延迟每千次调用成本结构化输出支持模型A92%1.2s较高原生支持模型B88%0.8s中等需Prompt约束模型C85%2.5s较低原生支持模型D90%1.5s低需Prompt约束最终我选的是模型B因为它的延迟最低准确率虽然比A低4个百分点但通过优化Prompt和加一层后处理实际效果差距缩小到了1个百分点以内而成本只有A的三分之一。2.3 向量数据库的选型逻辑与踩坑记录向量数据库是RAG架构的核心组件选型的时候要考虑数据规模、查询性能、运维成本、生态兼容性。我先后用过Milvus、Qdrant、Chroma、Pgvector这几种每个都有自己的适用场景。Milvus功能最全支持多种索引类型和分布式部署适合数据量在千万级别以上的场景。但它的运维复杂度也最高光是部署一套可用的集群就要折腾好几天。Qdrant的平衡性最好单机性能强支持过滤和混合检索API设计也很友好是我目前最推荐的方案。Chroma适合快速原型验证轻量级几行代码就能跑起来但不适合生产环境。Pgvector最大的优势是如果你已经在用PostgreSQL可以直接复用现有的数据库不用额外维护一套系统但性能上限相对较低。这里有个坑我必须提醒向量维度不是越高越好。我早期用某个嵌入模型输出是3072维的向量存了100万条数据内存直接爆了。后来换成768维的模型检索效果几乎没有下降但内存占用降到了原来的四分之一。所以选嵌入模型的时候一定要在效果和资源消耗之间做权衡。3. 核心模块的细节拆解与实操要点3.1 文档处理流水线的设计细节RAG系统的效果好不好七成取决于文档处理的质量。很多人把文档直接扔进去做嵌入结果检索出来的内容驴唇不对马嘴。问题出在文档处理这一步没做好。完整的文档处理流水线应该包括格式解析、内容清洗、分块、元数据提取、嵌入生成、索引构建。格式解析就是把PDF、Word、HTML这些不同格式的文件转成纯文本。这一步的坑在于PDF特别是扫描版的PDF需要OCR才能提取文字。我推荐用PyMuPDF做PDF解析速度快对表格和图片的处理也比较好。内容清洗是去掉无关的字符、页眉页脚、重复内容。这一步看起来简单但实际很繁琐。我的做法是先用正则表达式做一轮粗清洗然后用规则引擎处理特殊情况。比如有些PDF每页都有页脚你需要识别出这种模式并去掉。分块是最关键的环节。分块大小直接影响检索效果。块太小上下文不完整模型没法生成好的回答块太大检索精度下降而且会浪费Token。我的经验值是中文文本每块300到500字英文文本每块200到300词。但这不是绝对的要根据文档类型调整。技术文档可以小一点因为概念比较密集叙述性文档可以大一点因为需要更多上下文才能理解。分块策略也有讲究。最简单的固定长度分块效果往往不好因为它会在句子中间切断。我推荐用递归分块先按段落分段落太长再按句子分句子还太长才按字符分。LangChain的RecursiveCharacterTextSplitter就是干这个的用起来很方便。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ], length_functionlen, ) chunks splitter.split_text(document_text)chunk_overlap这个参数很重要它让相邻的块之间有重叠内容避免关键信息刚好落在边界上被切断。一般设置成块大小的10%到20%就行。实操心得分块之后一定要人工抽查一批数据看看块的内容是否完整、是否有意义。我每次做新项目都会随机抽20个块出来看这一步能发现80%的问题。3.2 嵌入模型的本地部署与性能优化嵌入模型负责把文本转成向量是RAG系统的核心组件之一。虽然可以用API但考虑到成本和数据隐私很多场景下本地部署更合适。本地部署嵌入模型硬件方面主要看显存。以BGE-large这个模型为例模型本身大概1.3GB推理的时候需要额外的显存来存中间计算结果总共大概需要2到3GB显存。如果没有GPU用CPU也能跑但速度会慢很多大概每秒只能处理几十条文本。部署方式我推荐用ONNX Runtime或者TensorRT来做推理加速。ONNX Runtime的兼容性更好安装也简单基本是pip install就能用。TensorRT的性能更强但配置起来麻烦一些需要根据具体的GPU型号做优化。from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(BAAI/bge-large-zh-v1.5) model model.to(cuda) texts [这是第一段文本, 这是第二段文本] embeddings model.encode(texts, normalize_embeddingsTrue, batch_size32) print(embeddings.shape)normalize_embeddingsTrue这个参数很重要它把向量归一化到单位长度这样计算余弦相似度的时候只需要做点积就行速度快很多。batch_size根据显存大小调整一般32到128之间。性能优化方面有几个实用的技巧。第一是批处理不要一条一条地嵌入攒一批一起处理吞吐量能提升好几倍。第二是用半精度浮点数也就是FP16显存占用减半速度也能提升30%左右效果几乎没影响。第三是如果文本量特别大可以考虑用更小的模型比如BGE-small效果下降不多但速度快很多。3.3 检索策略的进阶玩法基础的向量检索就是拿查询向量去数据库里找最相似的Top-K个结果。但实际用起来单纯的向量检索效果往往不够好需要配合其他策略。混合检索是目前效果最好的方案之一。它同时做向量检索和关键词检索然后把两边的结果融合。向量检索擅长语义匹配比如用户问“怎么退款”能匹配到“退货流程”这样的内容。关键词检索擅长精确匹配比如用户搜“订单号12345”能精确找到包含这个订单号的文档。融合的方法一般用RRF也就是倒数排名融合公式是RRF_score Σ(1 / (k rank_i))其中k是一个常数一般取60rank_i是文档在第i个检索结果中的排名。这个公式的好处是不需要归一化不同检索系统的分数直接基于排名做融合。重排序是另一个提升效果的关键步骤。向量检索拿到Top-50的结果后用一个重排序模型对这50个结果做精细打分然后取Top-5给大模型。重排序模型比嵌入模型大推理慢但只对少量候选做处理总体延迟增加不多效果提升很明显。常用的重排序模型有BGE-reranker系列部署方式和嵌入模型类似。from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-large, use_fp16True) query 怎么申请退款 passages [退款流程说明..., 订单查询方法..., 账户设置指南...] scores reranker.compute_score([[query, p] for p in passages]) sorted_results sorted(zip(passages, scores), keylambda x: x[1], reverseTrue)注意重排序模型的选择要和嵌入模型匹配。如果嵌入模型是多语言的重排序模型最好也是多语言的否则跨语言场景下效果会打折扣。3.4 Prompt工程的结构化设计方法Prompt工程不是玄学它有一套可以系统化学习的方法论。我见过太多人写Prompt就是凭感觉今天加一句“请仔细思考”明天加一句“你是一个专业的助手”效果不稳定也不知道为什么。我的做法是把Prompt拆成五个部分角色定义、任务描述、上下文信息、输出格式、约束条件。角色定义告诉模型它应该以什么身份来回答比如“你是一个资深的法律顾问”。任务描述说清楚具体要做什么比如“根据提供的法律条文回答用户的问题”。上下文信息就是RAG检索出来的相关内容。输出格式规定回答的结构比如“用JSON格式输出包含answer和confidence两个字段”。约束条件列出不能做的事情比如“如果上下文中没有相关信息回答‘我无法回答这个问题’不要编造”。这种结构化写法有几个好处。第一是可维护每个部分独立修改不会互相影响。第二是可测试你可以单独测试每个部分的效果。第三是可复用角色定义和输出格式这些部分可以在不同任务之间共享。PROMPT_TEMPLATE 你是一个专业的客服助手负责回答用户关于产品使用的问题。 ## 任务 根据下面提供的参考信息回答用户的问题。如果参考信息中没有相关内容请如实告知用户。 ## 参考信息 {context} ## 用户问题 {question} ## 输出要求 请按以下JSON格式输出 {{ answer: 你的回答, confidence: high/medium/low, source: 引用的参考信息编号 }} ## 约束 - 不要编造参考信息中不存在的内容 - 回答要简洁控制在200字以内 - 如果参考信息不足以回答问题confidence设为low 这个模板里有个细节值得说输出格式用了JSON而且给了示例。这是因为大模型对结构化输出的支持程度不一样有些模型原生支持JSON模式有些不支持。给一个明确的示例能大幅提高输出格式的准确率。4. 完整实操流程与关键环节实现4.1 环境搭建与依赖管理从零开始搭建AI工程环境第一步是把基础设施准备好。我推荐用Docker Compose来管理所有依赖服务这样环境一致性好换机器也能快速复现。需要的基础服务包括向量数据库Qdrant、缓存Redis、关系型数据库PostgreSQL、监控Prometheus Grafana。下面是一个精简版的docker-compose.ymlversion: 3.8 services: qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 volumes: - qdrant_data:/qdrant/storage redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:16-alpine environment: POSTGRES_DB: ai_engine POSTGRES_USER: admin POSTGRES_PASSWORD: your_password ports: - 5432:5432 volumes: - pg_data:/var/lib/postgresql/data volumes: qdrant_data: pg_data:Python环境我建议用uv来管理比pip快很多依赖解析也更靠谱。核心依赖包括fastapiWeb框架、sentence-transformers嵌入模型、qdrant-client向量数据库客户端、openai大模型调用、langchain编排框架可选。uv init ai-engineering uv add fastapi uvicorn sentence-transformers qdrant-client openai uv add --dev pytest pytest-asyncio实操心得依赖版本一定要锁定。AI领域的库更新非常快今天能跑的代码明天可能就因为某个依赖升级而报错。用uv.lock或者requirements.txt把版本固定下来能省很多麻烦。4.2 文档入库的完整实现文档入库是整个系统的数据基础流程是读取文件、解析内容、清洗文本、分块、生成嵌入、存入向量数据库。import fitz from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance import uuid class DocumentIngester: def __init__(self, qdrant_url: str, collection_name: str): self.client QdrantClient(urlqdrant_url) self.collection_name collection_name self.embedder SentenceTransformer(BAAI/bge-large-zh-v1.5) self.splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ], ) def parse_pdf(self, file_path: str) - str: doc fitz.open(file_path) text for page in doc: text page.get_text() doc.close() return text def clean_text(self, text: str) - str: import re text re.sub(r\s, , text) text re.sub(r第\s*\d\s*页, , text) return text.strip() def ingest(self, file_path: str, metadata: dict): raw_text self.parse_pdf(file_path) cleaned self.clean_text(raw_text) chunks self.splitter.split_text(cleaned) points [] for i, chunk in enumerate(chunks): vector self.embedder.encode(chunk, normalize_embeddingsTrue) points.append(PointStruct( idstr(uuid.uuid4()), vectorvector.tolist(), payload{ text: chunk, source: file_path, chunk_index: i, **metadata } )) self.client.upsert( collection_nameself.collection_name, pointspoints ) return len(points)这段代码有几个关键点。第一parse_pdf用的是PyMuPDF它对中文PDF的支持比较好。第二clean_text里去掉了页码因为页码对检索没有意义反而会干扰。第三每个块都存了source和chunk_index方便后续追溯来源。第四嵌入的时候用了normalize_embeddingsTrue这样后续计算相似度直接用点积就行。创建集合的时候要注意向量维度和距离度量方式client.create_collection( collection_namedocuments, vectors_configVectorParams( size1024, distanceDistance.COSINE ) )size要和嵌入模型的输出维度一致BGE-large是1024维。distance用COSINE因为我们已经归一化了向量COSINE和点积是等价的。4.3 检索链路的组装与调优检索链路负责根据用户查询找到最相关的文档片段。完整的链路是查询改写、向量检索、关键词检索、融合排序、重排序。查询改写这一步很多人会忽略但它对效果影响很大。用户的问题往往比较口语化比如“我买的东西坏了怎么办”直接拿这个去检索可能效果不好。查询改写的目的是把口语化的问题转成更适合检索的形式比如改成“商品损坏 退换货 流程”。可以用大模型来做改写也可以用规则。def rewrite_query(query: str, llm_client) - str: prompt f将下面的用户问题改写成适合文档检索的关键词形式只输出改写结果不要解释。 用户问题{query} 改写结果 response llm_client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0 ) return response.choices[0].message.content.strip()向量检索就是拿查询向量去Qdrant里搜def vector_search(query: str, client, embedder, top_k: int 20): query_vector embedder.encode(query, normalize_embeddingsTrue) results client.search( collection_namedocuments, query_vectorquery_vector.tolist(), limittop_k ) return [(r.payload[text], r.score) for r in results]关键词检索可以用Elasticsearch或者简单的BM25实现。如果不想引入额外组件Qdrant本身也支持全文检索可以在payload上建全文索引。融合排序用RRFdef rrf_fusion(vector_results, keyword_results, k60): scores {} for rank, (text, _) in enumerate(vector_results): scores[text] scores.get(text, 0) 1 / (k rank 1) for rank, (text, _) in enumerate(keyword_results): scores[text] scores.get(text, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)重排序用BGE-rerankerdef rerank(query: str, candidates: list, reranker, top_k: int 5): pairs [[query, text] for text, _ in candidates] scores reranker.compute_score(pairs) ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) return [text for (text, _), _ in ranked[:top_k]]整个链路串起来就是def retrieve(query: str, llm_client, qdrant_client, embedder, reranker): rewritten rewrite_query(query, llm_client) vector_results vector_search(rewritten, qdrant_client, embedder, top_k20) keyword_results keyword_search(rewritten, top_k20) fused rrf_fusion(vector_results, keyword_results) final rerank(query, fused[:30], reranker, top_k5) return final实操心得检索链路的每个环节都要单独评估。我一般会准备一个测试集包含问题和对应的正确答案然后分别测查询改写、向量检索、融合、重排序各环节的召回率。这样能精确定位瓶颈在哪里。4.4 生成模块的工程化封装生成模块负责把检索到的上下文和用户问题组装成Prompt调用大模型然后解析输出。工程化封装要考虑几个问题超时处理、重试机制、降级策略、输出解析。import asyncio from openai import AsyncOpenAI class GenerationService: def __init__(self, api_key: str, model: str gpt-4o-mini): self.client AsyncOpenAI(api_keyapi_key) self.model model async def generate(self, query: str, contexts: list, timeout: float 10.0): context_text \n\n.join( f[{i1}] {ctx} for i, ctx in enumerate(contexts) ) prompt PROMPT_TEMPLATE.format( contextcontext_text, questionquery ) for attempt in range(3): try: response await asyncio.wait_for( self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, response_format{type: json_object} ), timeouttimeout ) return self._parse_response(response.choices[0].message.content) except asyncio.TimeoutError: if attempt 2: return {answer: 服务暂时不可用请稍后重试, confidence: low} await asyncio.sleep(1 * (attempt 1)) except Exception as e: if attempt 2: raise await asyncio.sleep(1 * (attempt 1)) def _parse_response(self, content: str) - dict: import json try: return json.loads(content) except json.JSONDecodeError: return {answer: content, confidence: medium, source: }这段代码里有几个工程上的考量。第一用了异步调用因为大模型API的延迟比较高异步能提高并发能力。第二加了超时控制避免请求卡死。第三有重试机制但重试次数不超过3次避免雪崩。第四用了response_format{type: json_object}来强制JSON输出减少解析失败的概率。第五解析失败的时候有兜底逻辑不会直接报错。4.5 服务化部署与性能压测把上面这些模块组装成一个FastAPI服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): question: str top_k: int 5 class QueryResponse(BaseModel): answer: str confidence: str sources: list app.post(/query, response_modelQueryResponse) async def query_endpoint(req: QueryRequest): try: contexts retrieve(req.question, llm_client, qdrant_client, embedder, reranker) result await generation_service.generate(req.question, contexts) return QueryResponse( answerresult[answer], confidenceresult[confidence], sourcescontexts ) except Exception as e: raise HTTPException(status_code500, detailstr(e))部署的时候用uvicorn配合gunicorn做进程管理gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000-w 4表示4个worker进程一般设置成CPU核数的2倍。但要注意如果嵌入模型和重排序模型是加载在GPU上的多进程会导致显存不够这时候要么用单进程要么把模型推理单独拆成一个服务。性能压测用locust或者wrk。我一般会关注三个指标QPS、P95延迟、错误率。一个可用的AI问答服务QPS至少要能到10以上P95延迟控制在3秒以内错误率低于1%。from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(1, 3) task def query(self): self.client.post(/query, json{ question: 怎么申请退款, top_k: 5 })压测的时候要特别注意嵌入模型和重排序模型是计算密集型的并发一高就容易成为瓶颈。如果发现延迟飙升优先考虑给这两个模块加缓存相同查询直接返回缓存结果。5. 常见问题排查与避坑指南5.1 检索效果差的排查思路检索效果差是最常见的问题表现是模型回答的内容和问题不相关或者回答“我无法回答这个问题”但文档里明明有答案。排查要按链路一步步来。第一步检查分块质量。随机抽20个块出来看如果块的内容不完整、被截断或者包含大量无关内容那就是分块策略有问题。调整chunk_size和chunk_overlap或者换一种分块方式。第二步检查嵌入模型是否适合你的领域。通用嵌入模型在专业领域比如医疗、法律效果可能不好。可以试试在领域数据上微调嵌入模型或者换一个在该领域表现更好的模型。第三步检查检索参数。top_k太小可能漏掉相关文档太大则引入噪声。一般向量检索取20到50重排序后取3到5。相似度阈值也要调太低会引入不相关的结果太高会漏掉相关的结果。第四步检查查询改写。如果改写后的查询偏离了原意检索效果肯定差。可以先把查询改写关掉直接用原始查询检索对比效果。下面这个表格是我总结的检索问题速查表现象可能原因排查方法解决方案回答不相关分块太大或太小抽查分块内容调整chunk_size漏掉相关文档top_k太小增大top_k看召回增大top_k或加关键词检索检索结果重复分块重叠过多检查chunk_overlap减小overlap专业术语匹配差嵌入模型领域不匹配换模型测试微调或换领域模型查询改写偏离改写Prompt有问题关闭改写对比优化改写Prompt5.2 大模型输出不稳定的处理大模型输出不稳定是另一个高频问题。同样的输入有时候输出JSON格式正确有时候多了一段解释文字有时候字段名拼错了。这个问题在工程上必须解决否则下游解析会频繁报错。我的处理策略是三层防护。第一层是Prompt层面用明确的格式说明和示例加上response_format参数强制JSON输出。第二层是解析层面用宽容的解析器能处理常见的格式偏差比如去掉markdown代码块标记、修复缺失的引号。第三层是重试层面解析失败的时候自动重试重试时在Prompt里加上“上次输出格式有误请严格按照JSON格式输出”。def robust_json_parse(content: str) - dict: import json import re content content.strip() content re.sub(r^json\s*, , content) content re.sub(r\s*$, , content) try: return json.loads(content) except json.JSONDecodeError: pass match re.search(r\{.*\}, content, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return {answer: content, confidence: low, source: }注意不要指望大模型100%按格式输出工程上必须假设它会出错然后做好兜底。5.3 性能瓶颈的定位与优化性能问题在AI应用里特别突出因为链路上每个环节都可能成为瓶颈。定位性能问题要用工具不能靠猜。我一般用Python的cProfile做函数级性能分析用py-spy做线上环境的实时分析。常见的性能瓶颈和优化方案嵌入生成慢优化方法是批处理加GPU加速如果还慢就换更小的模型。向量检索慢优化方法是建索引Qdrant支持HNSW索引建好之后检索速度能提升10倍以上。大模型调用慢优化方法是流式输出加缓存相同问题直接返回缓存结果。重排序慢优化方法是减少候选数量从50降到20效果影响不大但速度翻倍。缓存是提升性能最有效的手段之一。我在Redis里缓存了两类数据嵌入向量和最终回答。嵌入向量缓存的key是文本的哈希值回答缓存的key是问题的哈希值。缓存命中率在真实场景下能到30%到50%效果非常明显。import hashlib import json def get_cache_key(text: str) - str: return fcache:{hashlib.md5(text.encode()).hexdigest()} async def get_or_compute(redis_client, key: str, compute_func, ttl: int 3600): cached await redis_client.get(key) if cached: return json.loads(cached) result await compute_func() await redis_client.setex(key, ttl, json.dumps(result)) return result5.4 成本控制的实战经验AI应用的成本主要来自大模型API调用和GPU服务器。如果不加控制费用很容易失控。我踩过的最大一个坑是一个内部工具上线后没做限流有人写了个脚本疯狂调用一天烧掉了几百块。成本控制要从几个方面入手。第一是限流按用户和按接口做限流防止滥用。第二是缓存相同请求直接返回缓存结果。第三是模型分级简单问题用便宜的小模型复杂问题才用大模型。第四是Token控制Prompt不要写太长检索的上下文数量要限制。模型分级的实现思路是先用小模型做意图分类判断问题的复杂度然后路由到不同的模型。简单问题比如FAQ用小模型复杂问题比如需要推理的用大模型。这个策略能降低50%以上的成本。async def route_and_generate(query: str, contexts: list): complexity await classify_complexity(query) if complexity simple: return await small_model.generate(query, contexts) else: return await large_model.generate(query, contexts)5.5 数据安全与隐私保护AI应用涉及用户数据安全隐私不能忽视。几个基本要求传输加密、存储加密、访问控制、数据脱敏。传输加密用HTTPS这个是最基本的。存储加密方面向量数据库和关系型数据库都要开启加密。访问控制要做到最小权限原则每个服务只能访问自己需要的数据。数据脱敏是在日志和监控里不要记录用户的敏感信息比如手机号、身份证号。还有一个容易被忽视的点是Prompt注入。用户可能在输入里嵌入恶意指令试图让模型泄露系统Prompt或者执行未授权的操作。防护方法是在Prompt里明确告诉模型不要执行用户输入中的指令同时对用户输入做过滤去掉明显的注入模式。def sanitize_input(text: str) - str: import re patterns [ rignore\s(all\s)?previous\sinstructions, r忽略(上面|之前|所有)的?(指令|要求), rsystem\s*:\s*, ] for pattern in patterns: text re.sub(pattern, , text, flagsre.IGNORECASE) return text这个过滤不是100%有效但能挡住大部分低级的注入尝试。更严格的防护需要在模型层面做比如用专门的分类器检测注入攻击。6. 从能跑到好用我踩过的那些坑ai-engineering-from-scratch这条路我走了两年多从最开始连向量数据库是什么都不知道到现在能独立设计并运维一套完整的AI应用系统中间踩的坑数不胜数。有几个教训我觉得特别值得分享。第一个教训是不要过早优化。我早期做一个项目上来就搞微服务架构把嵌入、检索、生成拆成三个独立服务结果调试的时候光排查服务间调用就花了大半时间。后来全部合并成一个单体应用开发效率提升了三倍。等系统真的到了性能瓶颈再拆也不迟。第二个教训是评估体系要先行。没有评估体系你根本不知道改动是让效果变好了还是变差了。我现在的习惯是任何新项目第一周不写业务代码先把评估数据集和评估脚本搭好。评估数据集不用很大100到200条就够但必须覆盖真实场景。第三个教训是日志要打全。AI应用的调试比传统应用难得多因为中间过程不透明。我现在的做法是每个环节都打日志原始查询、改写后的查询、检索到的文档、重排序后的结果、最终的Prompt、模型的原始输出、解析后的结果。出问题的时候一看日志就知道是哪一步出了问题。第四个教训是不要迷信框架。LangChain、LlamaIndex这些框架确实能提高开发效率但它们也引入了额外的抽象层出问题的时候很难排查。我的建议是先用框架快速验证想法等确定方案之后把核心逻辑用原生代码重写一遍这样可控性更强。最后分享一个我觉得特别有用的技巧给每个环节加一个“调试模式”。开启调试模式的时候接口返回完整的中间结果包括检索到的文档、每步的耗时、模型的原始输出。这个功能在排查问题和给团队演示的时候特别有用。这套东西后续还可以往几个方向扩展。一是加多模态能力支持图片和表格的检索。二是加Agent能力让系统能调用外部工具。三是加反馈闭环根据用户的点赞点踩自动优化检索和生成策略。每个方向都值得单独写一篇后面有机会再展开聊。 从零搭建AI工程体系：分层架构、RAG与向量数据库实战指南
1. 从零搭建AI工程体系为什么我劝你别再东拼西凑这两年AI应用开发的门槛肉眼可见地降低了随便调个API就能跑出一个能对话的Demo但真正要把一个AI功能做成稳定、可维护、能扛住真实用户流量的工程系统绝大多数人卡在了同一个地方知识全是碎片。今天看了一篇讲RAG的帖子明天刷到一个讲向量数据库的视频后天又被人安利某个Agent框架结果真到自己动手的时候发现连一个完整的请求链路都串不起来。ai-engineering-from-scratch这个方向说白了就是解决这个问题的。它不是某一个具体的开源项目而是一套从底层原理到工程落地的完整学习与实践路径。核心目标是让你具备独立设计、开发、部署、运维一个AI应用系统的能力而不是停留在“会调API”的层面。这套东西适合谁我认为有三类人最该认真对待一是后端或全栈工程师想转型做AI应用开发二是有一定Python基础但没系统做过AI工程的数据分析人员三是技术负责人需要评估AI项目的可行性和技术选型。我自己的经历比较典型。两年前我开始接触大模型应用开发那时候市面上还没有现在这么多成熟的框架踩的坑一个接一个。最惨的一次是做一个文档问答系统本地测试效果很好上线之后并发一上来响应时间直接从2秒飙到30秒用户投诉不断。排查了两天才发现是向量检索那块没有做索引优化每次查询都在做全量暴力搜索。这件事让我意识到AI工程和传统后端工程虽然有交集但有很多独特的坑必须系统性地去理解和掌握。这篇文章我会把ai-engineering-from-scratch涉及的核心环节全部拆开讲从整体架构设计思路到每个模块的具体实现细节再到实操过程中会遇到的问题和排查方法。内容会比较长但如果你真的想把这套东西搞明白建议耐心看完。我会尽量用大白话解释每个技术选择背后的原因让你不仅知道怎么做更知道为什么这么做。2. 整体架构设计与技术选型思路2.1 为什么分层架构是AI工程的起点很多人做AI应用上来就开始写Prompt、调模型代码全堆在一个文件里。这种写法做Demo没问题但一旦需求变化或者要加新功能整个代码就会变成一团乱麻。我在早期项目里就吃过这个亏一个客服机器人项目最开始只有问答功能后来要加意图识别、多轮对话管理、知识库检索代码从200行膨胀到3000多行改一个地方要动全身。从零搭建AI工程体系第一件事就是建立清晰的分层架构。我推荐的分层方式是四层接入层、编排层、能力层、基础设施层。接入层负责处理用户请求的接收和响应的返回包括API网关、鉴权、限流这些传统后端的东西。编排层是整个系统的核心负责协调各个AI能力模块的调用顺序和逻辑比如先做意图识别再决定是走知识库检索还是直接调用模型生成。能力层是具体的AI功能实现包括大模型调用、向量检索、文本嵌入、重排序等等。基础设施层则是支撑上面所有功能的基础组件包括向量数据库、缓存、消息队列、监控系统。这样分层的好处是每一层可以独立演进。比如你想把底层的大模型从A换成B只需要改能力层的模型调用模块编排层和接入层完全不用动。再比如你想给系统加一个缓存机制也只需要在基础设施层加一个Redis然后在编排层里调用就行。注意分层不是目的解耦才是。不要为了分层而分层如果你的项目确实很简单一个单体应用就够了过度设计反而增加维护成本。2.2 模型选型的核心考量维度模型选型是AI工程里最容易被忽视但又极其关键的决策。我见过太多团队在这个问题上要么过于随意要么过于纠结。过于随意的是直接用最贵的模型觉得贵的就是好的过于纠结的是花几周时间做各种评测结果项目进度严重滞后。我的经验是模型选型要从四个维度来考量能力、成本、延迟、可控性。能力指的是模型在你具体任务上的表现这个必须用你自己的数据来测不能只看公开榜单。成本包括API调用费用和自部署的硬件成本要算清楚每千次请求的实际花费。延迟直接影响用户体验特别是需要实时响应的场景比如对话系统。可控性指的是你能不能对模型的行为做精细控制比如是否支持Function Calling、是否支持结构化输出、是否支持微调。具体怎么选我一般会准备一个包含50到100条真实场景数据的测试集然后拿几个候选模型跑一遍从准确率、响应时间、费用三个维度打分。下面这个表格是我最近一个项目里的实际评测结果供参考模型任务准确率平均延迟每千次调用成本结构化输出支持模型A92%1.2s较高原生支持模型B88%0.8s中等需Prompt约束模型C85%2.5s较低原生支持模型D90%1.5s低需Prompt约束最终我选的是模型B因为它的延迟最低准确率虽然比A低4个百分点但通过优化Prompt和加一层后处理实际效果差距缩小到了1个百分点以内而成本只有A的三分之一。2.3 向量数据库的选型逻辑与踩坑记录向量数据库是RAG架构的核心组件选型的时候要考虑数据规模、查询性能、运维成本、生态兼容性。我先后用过Milvus、Qdrant、Chroma、Pgvector这几种每个都有自己的适用场景。Milvus功能最全支持多种索引类型和分布式部署适合数据量在千万级别以上的场景。但它的运维复杂度也最高光是部署一套可用的集群就要折腾好几天。Qdrant的平衡性最好单机性能强支持过滤和混合检索API设计也很友好是我目前最推荐的方案。Chroma适合快速原型验证轻量级几行代码就能跑起来但不适合生产环境。Pgvector最大的优势是如果你已经在用PostgreSQL可以直接复用现有的数据库不用额外维护一套系统但性能上限相对较低。这里有个坑我必须提醒向量维度不是越高越好。我早期用某个嵌入模型输出是3072维的向量存了100万条数据内存直接爆了。后来换成768维的模型检索效果几乎没有下降但内存占用降到了原来的四分之一。所以选嵌入模型的时候一定要在效果和资源消耗之间做权衡。3. 核心模块的细节拆解与实操要点3.1 文档处理流水线的设计细节RAG系统的效果好不好七成取决于文档处理的质量。很多人把文档直接扔进去做嵌入结果检索出来的内容驴唇不对马嘴。问题出在文档处理这一步没做好。完整的文档处理流水线应该包括格式解析、内容清洗、分块、元数据提取、嵌入生成、索引构建。格式解析就是把PDF、Word、HTML这些不同格式的文件转成纯文本。这一步的坑在于PDF特别是扫描版的PDF需要OCR才能提取文字。我推荐用PyMuPDF做PDF解析速度快对表格和图片的处理也比较好。内容清洗是去掉无关的字符、页眉页脚、重复内容。这一步看起来简单但实际很繁琐。我的做法是先用正则表达式做一轮粗清洗然后用规则引擎处理特殊情况。比如有些PDF每页都有页脚你需要识别出这种模式并去掉。分块是最关键的环节。分块大小直接影响检索效果。块太小上下文不完整模型没法生成好的回答块太大检索精度下降而且会浪费Token。我的经验值是中文文本每块300到500字英文文本每块200到300词。但这不是绝对的要根据文档类型调整。技术文档可以小一点因为概念比较密集叙述性文档可以大一点因为需要更多上下文才能理解。分块策略也有讲究。最简单的固定长度分块效果往往不好因为它会在句子中间切断。我推荐用递归分块先按段落分段落太长再按句子分句子还太长才按字符分。LangChain的RecursiveCharacterTextSplitter就是干这个的用起来很方便。from langchain.text_splitter import RecursiveCharacterTextSplitter splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ], length_functionlen, ) chunks splitter.split_text(document_text)chunk_overlap这个参数很重要它让相邻的块之间有重叠内容避免关键信息刚好落在边界上被切断。一般设置成块大小的10%到20%就行。实操心得分块之后一定要人工抽查一批数据看看块的内容是否完整、是否有意义。我每次做新项目都会随机抽20个块出来看这一步能发现80%的问题。3.2 嵌入模型的本地部署与性能优化嵌入模型负责把文本转成向量是RAG系统的核心组件之一。虽然可以用API但考虑到成本和数据隐私很多场景下本地部署更合适。本地部署嵌入模型硬件方面主要看显存。以BGE-large这个模型为例模型本身大概1.3GB推理的时候需要额外的显存来存中间计算结果总共大概需要2到3GB显存。如果没有GPU用CPU也能跑但速度会慢很多大概每秒只能处理几十条文本。部署方式我推荐用ONNX Runtime或者TensorRT来做推理加速。ONNX Runtime的兼容性更好安装也简单基本是pip install就能用。TensorRT的性能更强但配置起来麻烦一些需要根据具体的GPU型号做优化。from sentence_transformers import SentenceTransformer import numpy as np model SentenceTransformer(BAAI/bge-large-zh-v1.5) model model.to(cuda) texts [这是第一段文本, 这是第二段文本] embeddings model.encode(texts, normalize_embeddingsTrue, batch_size32) print(embeddings.shape)normalize_embeddingsTrue这个参数很重要它把向量归一化到单位长度这样计算余弦相似度的时候只需要做点积就行速度快很多。batch_size根据显存大小调整一般32到128之间。性能优化方面有几个实用的技巧。第一是批处理不要一条一条地嵌入攒一批一起处理吞吐量能提升好几倍。第二是用半精度浮点数也就是FP16显存占用减半速度也能提升30%左右效果几乎没影响。第三是如果文本量特别大可以考虑用更小的模型比如BGE-small效果下降不多但速度快很多。3.3 检索策略的进阶玩法基础的向量检索就是拿查询向量去数据库里找最相似的Top-K个结果。但实际用起来单纯的向量检索效果往往不够好需要配合其他策略。混合检索是目前效果最好的方案之一。它同时做向量检索和关键词检索然后把两边的结果融合。向量检索擅长语义匹配比如用户问“怎么退款”能匹配到“退货流程”这样的内容。关键词检索擅长精确匹配比如用户搜“订单号12345”能精确找到包含这个订单号的文档。融合的方法一般用RRF也就是倒数排名融合公式是RRF_score Σ(1 / (k rank_i))其中k是一个常数一般取60rank_i是文档在第i个检索结果中的排名。这个公式的好处是不需要归一化不同检索系统的分数直接基于排名做融合。重排序是另一个提升效果的关键步骤。向量检索拿到Top-50的结果后用一个重排序模型对这50个结果做精细打分然后取Top-5给大模型。重排序模型比嵌入模型大推理慢但只对少量候选做处理总体延迟增加不多效果提升很明显。常用的重排序模型有BGE-reranker系列部署方式和嵌入模型类似。from FlagEmbedding import FlagReranker reranker FlagReranker(BAAI/bge-reranker-large, use_fp16True) query 怎么申请退款 passages [退款流程说明..., 订单查询方法..., 账户设置指南...] scores reranker.compute_score([[query, p] for p in passages]) sorted_results sorted(zip(passages, scores), keylambda x: x[1], reverseTrue)注意重排序模型的选择要和嵌入模型匹配。如果嵌入模型是多语言的重排序模型最好也是多语言的否则跨语言场景下效果会打折扣。3.4 Prompt工程的结构化设计方法Prompt工程不是玄学它有一套可以系统化学习的方法论。我见过太多人写Prompt就是凭感觉今天加一句“请仔细思考”明天加一句“你是一个专业的助手”效果不稳定也不知道为什么。我的做法是把Prompt拆成五个部分角色定义、任务描述、上下文信息、输出格式、约束条件。角色定义告诉模型它应该以什么身份来回答比如“你是一个资深的法律顾问”。任务描述说清楚具体要做什么比如“根据提供的法律条文回答用户的问题”。上下文信息就是RAG检索出来的相关内容。输出格式规定回答的结构比如“用JSON格式输出包含answer和confidence两个字段”。约束条件列出不能做的事情比如“如果上下文中没有相关信息回答‘我无法回答这个问题’不要编造”。这种结构化写法有几个好处。第一是可维护每个部分独立修改不会互相影响。第二是可测试你可以单独测试每个部分的效果。第三是可复用角色定义和输出格式这些部分可以在不同任务之间共享。PROMPT_TEMPLATE 你是一个专业的客服助手负责回答用户关于产品使用的问题。 ## 任务 根据下面提供的参考信息回答用户的问题。如果参考信息中没有相关内容请如实告知用户。 ## 参考信息 {context} ## 用户问题 {question} ## 输出要求 请按以下JSON格式输出 {{ answer: 你的回答, confidence: high/medium/low, source: 引用的参考信息编号 }} ## 约束 - 不要编造参考信息中不存在的内容 - 回答要简洁控制在200字以内 - 如果参考信息不足以回答问题confidence设为low 这个模板里有个细节值得说输出格式用了JSON而且给了示例。这是因为大模型对结构化输出的支持程度不一样有些模型原生支持JSON模式有些不支持。给一个明确的示例能大幅提高输出格式的准确率。4. 完整实操流程与关键环节实现4.1 环境搭建与依赖管理从零开始搭建AI工程环境第一步是把基础设施准备好。我推荐用Docker Compose来管理所有依赖服务这样环境一致性好换机器也能快速复现。需要的基础服务包括向量数据库Qdrant、缓存Redis、关系型数据库PostgreSQL、监控Prometheus Grafana。下面是一个精简版的docker-compose.ymlversion: 3.8 services: qdrant: image: qdrant/qdrant:latest ports: - 6333:6333 volumes: - qdrant_data:/qdrant/storage redis: image: redis:7-alpine ports: - 6379:6379 postgres: image: postgres:16-alpine environment: POSTGRES_DB: ai_engine POSTGRES_USER: admin POSTGRES_PASSWORD: your_password ports: - 5432:5432 volumes: - pg_data:/var/lib/postgresql/data volumes: qdrant_data: pg_data:Python环境我建议用uv来管理比pip快很多依赖解析也更靠谱。核心依赖包括fastapiWeb框架、sentence-transformers嵌入模型、qdrant-client向量数据库客户端、openai大模型调用、langchain编排框架可选。uv init ai-engineering uv add fastapi uvicorn sentence-transformers qdrant-client openai uv add --dev pytest pytest-asyncio实操心得依赖版本一定要锁定。AI领域的库更新非常快今天能跑的代码明天可能就因为某个依赖升级而报错。用uv.lock或者requirements.txt把版本固定下来能省很多麻烦。4.2 文档入库的完整实现文档入库是整个系统的数据基础流程是读取文件、解析内容、清洗文本、分块、生成嵌入、存入向量数据库。import fitz from langchain.text_splitter import RecursiveCharacterTextSplitter from sentence_transformers import SentenceTransformer from qdrant_client import QdrantClient from qdrant_client.models import PointStruct, VectorParams, Distance import uuid class DocumentIngester: def __init__(self, qdrant_url: str, collection_name: str): self.client QdrantClient(urlqdrant_url) self.collection_name collection_name self.embedder SentenceTransformer(BAAI/bge-large-zh-v1.5) self.splitter RecursiveCharacterTextSplitter( chunk_size500, chunk_overlap50, separators[\n\n, \n, 。, , , , , ], ) def parse_pdf(self, file_path: str) - str: doc fitz.open(file_path) text for page in doc: text page.get_text() doc.close() return text def clean_text(self, text: str) - str: import re text re.sub(r\s, , text) text re.sub(r第\s*\d\s*页, , text) return text.strip() def ingest(self, file_path: str, metadata: dict): raw_text self.parse_pdf(file_path) cleaned self.clean_text(raw_text) chunks self.splitter.split_text(cleaned) points [] for i, chunk in enumerate(chunks): vector self.embedder.encode(chunk, normalize_embeddingsTrue) points.append(PointStruct( idstr(uuid.uuid4()), vectorvector.tolist(), payload{ text: chunk, source: file_path, chunk_index: i, **metadata } )) self.client.upsert( collection_nameself.collection_name, pointspoints ) return len(points)这段代码有几个关键点。第一parse_pdf用的是PyMuPDF它对中文PDF的支持比较好。第二clean_text里去掉了页码因为页码对检索没有意义反而会干扰。第三每个块都存了source和chunk_index方便后续追溯来源。第四嵌入的时候用了normalize_embeddingsTrue这样后续计算相似度直接用点积就行。创建集合的时候要注意向量维度和距离度量方式client.create_collection( collection_namedocuments, vectors_configVectorParams( size1024, distanceDistance.COSINE ) )size要和嵌入模型的输出维度一致BGE-large是1024维。distance用COSINE因为我们已经归一化了向量COSINE和点积是等价的。4.3 检索链路的组装与调优检索链路负责根据用户查询找到最相关的文档片段。完整的链路是查询改写、向量检索、关键词检索、融合排序、重排序。查询改写这一步很多人会忽略但它对效果影响很大。用户的问题往往比较口语化比如“我买的东西坏了怎么办”直接拿这个去检索可能效果不好。查询改写的目的是把口语化的问题转成更适合检索的形式比如改成“商品损坏 退换货 流程”。可以用大模型来做改写也可以用规则。def rewrite_query(query: str, llm_client) - str: prompt f将下面的用户问题改写成适合文档检索的关键词形式只输出改写结果不要解释。 用户问题{query} 改写结果 response llm_client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: prompt}], temperature0 ) return response.choices[0].message.content.strip()向量检索就是拿查询向量去Qdrant里搜def vector_search(query: str, client, embedder, top_k: int 20): query_vector embedder.encode(query, normalize_embeddingsTrue) results client.search( collection_namedocuments, query_vectorquery_vector.tolist(), limittop_k ) return [(r.payload[text], r.score) for r in results]关键词检索可以用Elasticsearch或者简单的BM25实现。如果不想引入额外组件Qdrant本身也支持全文检索可以在payload上建全文索引。融合排序用RRFdef rrf_fusion(vector_results, keyword_results, k60): scores {} for rank, (text, _) in enumerate(vector_results): scores[text] scores.get(text, 0) 1 / (k rank 1) for rank, (text, _) in enumerate(keyword_results): scores[text] scores.get(text, 0) 1 / (k rank 1) return sorted(scores.items(), keylambda x: x[1], reverseTrue)重排序用BGE-rerankerdef rerank(query: str, candidates: list, reranker, top_k: int 5): pairs [[query, text] for text, _ in candidates] scores reranker.compute_score(pairs) ranked sorted(zip(candidates, scores), keylambda x: x[1], reverseTrue) return [text for (text, _), _ in ranked[:top_k]]整个链路串起来就是def retrieve(query: str, llm_client, qdrant_client, embedder, reranker): rewritten rewrite_query(query, llm_client) vector_results vector_search(rewritten, qdrant_client, embedder, top_k20) keyword_results keyword_search(rewritten, top_k20) fused rrf_fusion(vector_results, keyword_results) final rerank(query, fused[:30], reranker, top_k5) return final实操心得检索链路的每个环节都要单独评估。我一般会准备一个测试集包含问题和对应的正确答案然后分别测查询改写、向量检索、融合、重排序各环节的召回率。这样能精确定位瓶颈在哪里。4.4 生成模块的工程化封装生成模块负责把检索到的上下文和用户问题组装成Prompt调用大模型然后解析输出。工程化封装要考虑几个问题超时处理、重试机制、降级策略、输出解析。import asyncio from openai import AsyncOpenAI class GenerationService: def __init__(self, api_key: str, model: str gpt-4o-mini): self.client AsyncOpenAI(api_keyapi_key) self.model model async def generate(self, query: str, contexts: list, timeout: float 10.0): context_text \n\n.join( f[{i1}] {ctx} for i, ctx in enumerate(contexts) ) prompt PROMPT_TEMPLATE.format( contextcontext_text, questionquery ) for attempt in range(3): try: response await asyncio.wait_for( self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, response_format{type: json_object} ), timeouttimeout ) return self._parse_response(response.choices[0].message.content) except asyncio.TimeoutError: if attempt 2: return {answer: 服务暂时不可用请稍后重试, confidence: low} await asyncio.sleep(1 * (attempt 1)) except Exception as e: if attempt 2: raise await asyncio.sleep(1 * (attempt 1)) def _parse_response(self, content: str) - dict: import json try: return json.loads(content) except json.JSONDecodeError: return {answer: content, confidence: medium, source: }这段代码里有几个工程上的考量。第一用了异步调用因为大模型API的延迟比较高异步能提高并发能力。第二加了超时控制避免请求卡死。第三有重试机制但重试次数不超过3次避免雪崩。第四用了response_format{type: json_object}来强制JSON输出减少解析失败的概率。第五解析失败的时候有兜底逻辑不会直接报错。4.5 服务化部署与性能压测把上面这些模块组装成一个FastAPI服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel app FastAPI() class QueryRequest(BaseModel): question: str top_k: int 5 class QueryResponse(BaseModel): answer: str confidence: str sources: list app.post(/query, response_modelQueryResponse) async def query_endpoint(req: QueryRequest): try: contexts retrieve(req.question, llm_client, qdrant_client, embedder, reranker) result await generation_service.generate(req.question, contexts) return QueryResponse( answerresult[answer], confidenceresult[confidence], sourcescontexts ) except Exception as e: raise HTTPException(status_code500, detailstr(e))部署的时候用uvicorn配合gunicorn做进程管理gunicorn main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000-w 4表示4个worker进程一般设置成CPU核数的2倍。但要注意如果嵌入模型和重排序模型是加载在GPU上的多进程会导致显存不够这时候要么用单进程要么把模型推理单独拆成一个服务。性能压测用locust或者wrk。我一般会关注三个指标QPS、P95延迟、错误率。一个可用的AI问答服务QPS至少要能到10以上P95延迟控制在3秒以内错误率低于1%。from locust import HttpUser, task, between class AIUser(HttpUser): wait_time between(1, 3) task def query(self): self.client.post(/query, json{ question: 怎么申请退款, top_k: 5 })压测的时候要特别注意嵌入模型和重排序模型是计算密集型的并发一高就容易成为瓶颈。如果发现延迟飙升优先考虑给这两个模块加缓存相同查询直接返回缓存结果。5. 常见问题排查与避坑指南5.1 检索效果差的排查思路检索效果差是最常见的问题表现是模型回答的内容和问题不相关或者回答“我无法回答这个问题”但文档里明明有答案。排查要按链路一步步来。第一步检查分块质量。随机抽20个块出来看如果块的内容不完整、被截断或者包含大量无关内容那就是分块策略有问题。调整chunk_size和chunk_overlap或者换一种分块方式。第二步检查嵌入模型是否适合你的领域。通用嵌入模型在专业领域比如医疗、法律效果可能不好。可以试试在领域数据上微调嵌入模型或者换一个在该领域表现更好的模型。第三步检查检索参数。top_k太小可能漏掉相关文档太大则引入噪声。一般向量检索取20到50重排序后取3到5。相似度阈值也要调太低会引入不相关的结果太高会漏掉相关的结果。第四步检查查询改写。如果改写后的查询偏离了原意检索效果肯定差。可以先把查询改写关掉直接用原始查询检索对比效果。下面这个表格是我总结的检索问题速查表现象可能原因排查方法解决方案回答不相关分块太大或太小抽查分块内容调整chunk_size漏掉相关文档top_k太小增大top_k看召回增大top_k或加关键词检索检索结果重复分块重叠过多检查chunk_overlap减小overlap专业术语匹配差嵌入模型领域不匹配换模型测试微调或换领域模型查询改写偏离改写Prompt有问题关闭改写对比优化改写Prompt5.2 大模型输出不稳定的处理大模型输出不稳定是另一个高频问题。同样的输入有时候输出JSON格式正确有时候多了一段解释文字有时候字段名拼错了。这个问题在工程上必须解决否则下游解析会频繁报错。我的处理策略是三层防护。第一层是Prompt层面用明确的格式说明和示例加上response_format参数强制JSON输出。第二层是解析层面用宽容的解析器能处理常见的格式偏差比如去掉markdown代码块标记、修复缺失的引号。第三层是重试层面解析失败的时候自动重试重试时在Prompt里加上“上次输出格式有误请严格按照JSON格式输出”。def robust_json_parse(content: str) - dict: import json import re content content.strip() content re.sub(r^json\s*, , content) content re.sub(r\s*$, , content) try: return json.loads(content) except json.JSONDecodeError: pass match re.search(r\{.*\}, content, re.DOTALL) if match: try: return json.loads(match.group()) except json.JSONDecodeError: pass return {answer: content, confidence: low, source: }注意不要指望大模型100%按格式输出工程上必须假设它会出错然后做好兜底。5.3 性能瓶颈的定位与优化性能问题在AI应用里特别突出因为链路上每个环节都可能成为瓶颈。定位性能问题要用工具不能靠猜。我一般用Python的cProfile做函数级性能分析用py-spy做线上环境的实时分析。常见的性能瓶颈和优化方案嵌入生成慢优化方法是批处理加GPU加速如果还慢就换更小的模型。向量检索慢优化方法是建索引Qdrant支持HNSW索引建好之后检索速度能提升10倍以上。大模型调用慢优化方法是流式输出加缓存相同问题直接返回缓存结果。重排序慢优化方法是减少候选数量从50降到20效果影响不大但速度翻倍。缓存是提升性能最有效的手段之一。我在Redis里缓存了两类数据嵌入向量和最终回答。嵌入向量缓存的key是文本的哈希值回答缓存的key是问题的哈希值。缓存命中率在真实场景下能到30%到50%效果非常明显。import hashlib import json def get_cache_key(text: str) - str: return fcache:{hashlib.md5(text.encode()).hexdigest()} async def get_or_compute(redis_client, key: str, compute_func, ttl: int 3600): cached await redis_client.get(key) if cached: return json.loads(cached) result await compute_func() await redis_client.setex(key, ttl, json.dumps(result)) return result5.4 成本控制的实战经验AI应用的成本主要来自大模型API调用和GPU服务器。如果不加控制费用很容易失控。我踩过的最大一个坑是一个内部工具上线后没做限流有人写了个脚本疯狂调用一天烧掉了几百块。成本控制要从几个方面入手。第一是限流按用户和按接口做限流防止滥用。第二是缓存相同请求直接返回缓存结果。第三是模型分级简单问题用便宜的小模型复杂问题才用大模型。第四是Token控制Prompt不要写太长检索的上下文数量要限制。模型分级的实现思路是先用小模型做意图分类判断问题的复杂度然后路由到不同的模型。简单问题比如FAQ用小模型复杂问题比如需要推理的用大模型。这个策略能降低50%以上的成本。async def route_and_generate(query: str, contexts: list): complexity await classify_complexity(query) if complexity simple: return await small_model.generate(query, contexts) else: return await large_model.generate(query, contexts)5.5 数据安全与隐私保护AI应用涉及用户数据安全隐私不能忽视。几个基本要求传输加密、存储加密、访问控制、数据脱敏。传输加密用HTTPS这个是最基本的。存储加密方面向量数据库和关系型数据库都要开启加密。访问控制要做到最小权限原则每个服务只能访问自己需要的数据。数据脱敏是在日志和监控里不要记录用户的敏感信息比如手机号、身份证号。还有一个容易被忽视的点是Prompt注入。用户可能在输入里嵌入恶意指令试图让模型泄露系统Prompt或者执行未授权的操作。防护方法是在Prompt里明确告诉模型不要执行用户输入中的指令同时对用户输入做过滤去掉明显的注入模式。def sanitize_input(text: str) - str: import re patterns [ rignore\s(all\s)?previous\sinstructions, r忽略(上面|之前|所有)的?(指令|要求), rsystem\s*:\s*, ] for pattern in patterns: text re.sub(pattern, , text, flagsre.IGNORECASE) return text这个过滤不是100%有效但能挡住大部分低级的注入尝试。更严格的防护需要在模型层面做比如用专门的分类器检测注入攻击。6. 从能跑到好用我踩过的那些坑ai-engineering-from-scratch这条路我走了两年多从最开始连向量数据库是什么都不知道到现在能独立设计并运维一套完整的AI应用系统中间踩的坑数不胜数。有几个教训我觉得特别值得分享。第一个教训是不要过早优化。我早期做一个项目上来就搞微服务架构把嵌入、检索、生成拆成三个独立服务结果调试的时候光排查服务间调用就花了大半时间。后来全部合并成一个单体应用开发效率提升了三倍。等系统真的到了性能瓶颈再拆也不迟。第二个教训是评估体系要先行。没有评估体系你根本不知道改动是让效果变好了还是变差了。我现在的习惯是任何新项目第一周不写业务代码先把评估数据集和评估脚本搭好。评估数据集不用很大100到200条就够但必须覆盖真实场景。第三个教训是日志要打全。AI应用的调试比传统应用难得多因为中间过程不透明。我现在的做法是每个环节都打日志原始查询、改写后的查询、检索到的文档、重排序后的结果、最终的Prompt、模型的原始输出、解析后的结果。出问题的时候一看日志就知道是哪一步出了问题。第四个教训是不要迷信框架。LangChain、LlamaIndex这些框架确实能提高开发效率但它们也引入了额外的抽象层出问题的时候很难排查。我的建议是先用框架快速验证想法等确定方案之后把核心逻辑用原生代码重写一遍这样可控性更强。最后分享一个我觉得特别有用的技巧给每个环节加一个“调试模式”。开启调试模式的时候接口返回完整的中间结果包括检索到的文档、每步的耗时、模型的原始输出。这个功能在排查问题和给团队演示的时候特别有用。这套东西后续还可以往几个方向扩展。一是加多模态能力支持图片和表格的检索。二是加Agent能力让系统能调用外部工具。三是加反馈闭环根据用户的点赞点踩自动优化检索和生成策略。每个方向都值得单独写一篇后面有机会再展开聊。