项目ai-engineering-from-scratch——从零开始搭建AI工程。这个标题本身传递了两个关键信号第一它面向的是工程落地不是某个模型调参实验第二它强调from scratch意味着不依赖那些把所有东西封装好的一体化平台而是自己从底层把管线搭起来。这几年AI工程化话题持续升温但大量资料要么只讲调用API跑个demo要么一上来就丢给你一套重量级框架真正能让人理解每一层在干什么、为什么这么设计的资料反而稀缺。这篇就从我实际搭建的经验出发把从零构建AI工程的完整思路、代码骨架和踩坑记录分享出来。1. 整体设计与思路拆解1.1 为什么从零开始反而更值得做先泼一盆冷水除非你只做一次性验证否则别急着上那些号称全自动的AI应用搭建平台。它们的便捷是拿透明度换来的——出了问题你只能对着黑盒干瞪眼更别提深度定制时的处处掣肘。我见过好几个团队初期贪图快速度过demo阶段结果进入生产环境后因为无法自定义评估逻辑、无法精细控制模型调用频率和缓存策略被迫推倒重来。所谓ai-engineering核心能力不是会调API而是理解一条完整的AI链路模型怎么选、上下文怎么管理、工具怎么接入、结果怎么评估、性能怎么追踪。从零搭建的过程本质上是把这些环节逐个拆开、再亲手拼装起来。这个过程的产出不是一套只服务于单次任务的代码而是一个可扩展、可观测、可替换组件的工程底座。另一个现实原因是成本控制。商业平台的定价模型通常包含较高的溢价而且部分平台在模型版本更新、数据存储策略上的限制到了后期非常碍事。自己搭建核心链路后模型层可以随时切换不同的服务商或本地部署方案存储层可以用对象存储加向量数据库的组合替代平台自带方案这一层灵活性在长期迭代中价值巨大。1.2 核心架构的分层设计我最终落地的架构严格遵守了五层分离原则这里直接给出整体视图层级职责范围关键组件接入层对外提供API、处理鉴权与限流FastAPI、Redis编排层工作流调度、任务状态管理Temporal或Celery模型层统一封装各类模型调用、流式输出OpenAI SDK、LangChain回调检索层文档解析、向量化、相似度检索LlamaIndex、向量数据库观测层日志结构化、链路追踪、效果评估Langfuse、Prometheus这个分层的核心原则是上层依赖下层的接口而非实现。举个具体例子编排层触发一个问答任务时它只调用模型层定义的complete(messages)接口完全不需要知道这个接口背后是GPT-4o还是本地微调模型也不关心检索层用的向量数据库是Milvus还是Weaviate。这样设计带来的最直接的好处是每一层都可以独立演进。比如检索层初期用的开源嵌入模型效果不佳后期换成商用API只需修改检索层内部实现上层代码一行不动。我实际经历了一次完整的向量库迁移从FAISS迁到Milvus迁移过程中编排层和模型层零改动。1.3 依赖管理的取舍依赖管理是从零构建时最容易被低估的问题。很多人图省事直接把所有可能的库全部装进requirements.txt甚至用pip install langchain一把梭。但AI工程栈的特殊性在于每个库的依赖链很长版本之间经常互相冲突。比如某个版本的openai依赖httpx而新版的langchain又锁死了另一个版本的httpx冲突发生时你可能花费半天时间解决环境问题。我的建议是按层拆分依赖文件requirements-base.txt # 基础库pydantic、fastapi、redis requirements-model.txt # 模型层openai、anthropic、transformers requirements-retrieval.txt # 检索层llama-index、milvus-sdk、tiktoken requirements-dev.txt # 开发调试pytest、black、ruff、pre-commit每一个依赖文件在项目根目录下单独管理同时用pip-tools生成锁定版本文件确保团队内和CI环境的一致性。这个习惯在项目运行三个月后、依赖升级时会帮你省下大量痛苦。2. 核心细节解析与实操要点2.1 任务拆解与Prompt管理策略AI工程实践中Prompt的高效管理不是写几句提示词那么简单。它涉及版本管理、测试评估、变量注入、上下文组装等系统性工程问题。我见过很多团队的Prompt迭代方式直接改代码里的字符串改完上线出问题再回滚。这种方式在小规模时勉强能跑但当项目包含几十个Prompt模板、每种模板需要适配不同业务场景时代码里到处散落的字符串会成为灾难源。我采用的方案是Prompt即配置所有Prompt模板存放于独立目录使用Jinja2作为模板引擎通过结构化文件维护版本。{ name: rag_qa, version: 3.2.1, description: 面向RAG场景的问答主Prompt, variables: [context, question, history], template: 你是一名专业的{role}。请基于以下【参考资料】回答问题。\n\n【参考资料】\n{context}\n\n【用户问题】\n{question}\n\n要求\n1. 如果参考资料中没有相关信息请明确回答资料中未找到相关信息不要编造。\n2. 回答时请分点列出依据方便用户核对。 }实际加载模板的代码也非常简单from jinja2 import Environment, FileSystemLoader, StrictUndefined env Environment( loaderFileSystemLoader(prompts/), undefinedStrictUndefined, # 重要防止模板变量拼写错误被静默忽略 trim_blocksTrue, lstrip_blocksTrue ) def load_prompt(prompt_name: str, **variables) - str: template env.get_template(f{prompt_name}.json) # 实际这里要解析JSON并调用Jinja2渲染template字段 rendered env.from_string(template[template]).render(**variables) return renderedStrictUndefined这个设置是血泪教训换来的。默认情况下Jinja2对未定义的变量会渲染成空字符串这意味着如果变量名拼错你不会收到任何报错只会在生成结果中发现内容缺失。对于AI应用来说这种错误尤其隐蔽——它不会直接崩溃而是产生一个质量不达标的答案被用户当成AI能力不行误导后续排查方向。2.2 模型层统一封装的必要性模型层是AI工程的核心枢纽。它不只是一个简单的API调用封装还需要统一处理以下几个跨模型关注点超时与重试策略不同服务商的SLA不同网络抖动时需要有差异化的重试机制流式输出协议实时对话场景要求SSE流式返回需要统一封装为同步/异步迭代器Token计数与限流防止超出上下文窗口或预算超支内容安全过滤对输入输出进行敏感信息检测结构化输出确保模型返回结果可以被程序安全解析以下是我线上使用的一个简约但可靠的模型层封装import asyncio import json import time from typing import AsyncIterator, Optional from dataclasses import dataclass import openai dataclass class ModelConfig: model_name: str api_base: Optional[str] api_key_env: str max_tokens: int 1024 temperature: float 0.2 timeout: float 30.0 max_retries: int 3 class UnifiedLLMClient: def __init__(self, config: ModelConfig): self.config config # 动态读取API密钥不放硬编码 self.client openai.AsyncOpenAI( api_keyos.environ[config.api_key_env], max_retriesconfig.max_retries, timeoutconfig.timeout ) async def complete(self, messages: list[dict]) - dict: 非流式补全返回标准化结构 resp await self.client.chat.completions.create( modelself.config.model_name, messagesmessages, max_tokensself.config.max_tokens, temperatureself.config.temperature, ) usage resp.usage return { content: resp.choices[0].message.content, prompt_tokens: usage.prompt_tokens, completion_tokens: usage.completion_tokens, latency_ms: None # 由调用方计算 } async def stream(self, messages: list[dict]) - AsyncIterator[str]: 流式补全逐token产出文本 stream await self.client.chat.completions.create( modelself.config.model_name, messagesmessages, max_tokensself.config.max_tokens, temperatureself.config.temperature, streamTrue, ) async for chunk in stream: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content这里有几个工程细节值得说明API密钥不硬编码是安全底线一律通过环境变量或密钥管理服务注入max_tokens必须显式设置否则默认值可能过小导致输出被截断或者不同模型的默认值不一致导致行为不可预期返回结构统一不管底层模型原生的返回格式长什么样上层拿到的一定是content和usage两个字段构成的标准化字典2.3 上下文组装RAG管线的工程化要点RAG检索增强生成是AI工程最常见的落地形态。但把文档切一切、塞进向量库、检索出来拼到Prompt里这只是入门玩法。工程化的RAG管线需要考虑文档解析质量、切分策略、检索召回与重排、上下文压缩四个层面。文档解析是最容易被低估的环节。PDF解析有很多开源工具但实测下来不同类型PDF的解析效果差异非常大文本型PDF用pypdf就能解决扫描件需要OCR带复杂表格的文档需要专门的表格识别模型。我建议做一层抽象按文件类型动态选择解析器from typing import Protocol import fitz # PyMuPDF class DocParser(Protocol): def parse(self, file_path: str) - str: ... class PDFTextParser: 常规文本型PDF解析 def parse(self, file_path: str) - str: doc fitz.open(file_path) return \n.join(page.get_text() for page in doc) class PDFOCRParser: 扫描件PDF走OCR def parse(self, file_path: str) - str: doc fitz.open(file_path) texts [] for page in doc: pix page.get_pixmap(dpi200) img Image.open(io.BytesIO(pix.tobytes(png))) texts.append(ocr_engine(img)) return \n.join(texts)切分策略则是召回质量的关键。固定长度切分比如每500个字符切一段实现简单但会切断语义完整的段落导致检索到的片段信息不完整。我采用的策略是先按文档结构切分章节、段落再对过长的段落做递归切分同时保留标题作为前缀注入子片段。def recursive_split(text: str, chunk_size: int 500, overlap: int 50): 递归切分优先按段落段落过长按句子句子过长按字符 if len(text) chunk_size: return [text] # 优先尝试按段落切 paragraphs text.split(\n\n) if len(paragraphs) 1: chunks [] current for para in paragraphs: if len(current) len(para) 2 chunk_size: if current: chunks.append(current.strip()) current para else: current \n\n para if current: chunks.append(current.strip()) result [] for c in chunks: if len(c) chunk_size: result.extend(recursive_split(c, chunk_size, overlap)) else: result.append(c) return result # 段落太长按句子切 sentences text.replace(。, 。\n).replace(, \n).splitlines() # ... 类似逻辑展开切分的初衷是提升检索命中率。一个语义完整的检索单元比一段被硬生生截断的文字要有效得多。但也不要走向另一个极端——把整个文档作为一个检索单元召回粒度太粗噪声会干扰最终生成质量。2.4 评估体系AI工程的自测环节如果问AI工程和传统软件工程最大区别是什么我的答案是AI应用没有绝对正确的输出只有质量高低之分。这意味着测试环节不能简单断言输出是否等于期望值而需要一套多维度的质量评估体系。我在项目中落地了三层评估第一层是规则化检查用于检测确定性的错误。比如输出是否为空、是否包含指定格式、响应时间是否超限。第二层是模型辅助评估用一个更强的模型比如GPT-4o作为裁判对生成结果的多项指标打分async def llm_as_judge(output_text: str, reference: dict) - dict: judge_prompt 你是一个严格的内容质量评估员。请对以下AI生成结果打分1-5分 1. 相关性是否回答了用户问题 2. 忠实度是否有依据还是编造了内容 3. 完整性信息是否全面 4. 可读性表达是否清晰 用户请求{prompt} 生成结果{output} 请返回JSON格式{relevance: 4, faithfulness: 5, ...} # 调用评估模型并解析JSON结果 result await judge_client.complete(...) return json.loads(extract_json(result[content]))第三层是线上用户反馈在对外服务中显式提供有用/无用按钮并将反馈回流到评估数据集形成持续优化的闭环。这套体系的重要价值在模型升级时体现得尤为明显。每次切换底模我都先拿固定的100条评测集跑一遍评分对比观察分数的正负波动再决定是否切换。如果没有这套机制模型升级就是凭感觉拍板。3. 实操过程与核心环节实现3.1 项目结构与开发路线图实际动手时我建议按稳扎稳打的路径推进不要试图一口气搭建所有组件。这里给出我验证过的一条可行路线第一周搭建项目骨架、配置管理、日志系统基于structlog输出JSON结构化日志第二周实现模型层封装跑通单个模型的流式与非流式调用第三周搭建RAG检索层完成文档入库和基础问答第四周引入编排层实现多轮对话的状态管理和工具调用第五周完善评估体系和观测面板项目目录结构如下ai_engineering_scratch/ ├── app/ │ ├── api/ # FastAPI路由 │ │ ├── chat.py # 对话接口 │ │ └── health.py # 健康检查 │ ├── core/ # 配置、日志、异常处理 │ │ ├── config.py # Pydantic Settings配置模型 │ │ ├── logging.py # 结构化日志配置 │ │ └── exceptions.py # 统一异常定义 │ ├── model/ # 模型层 │ │ ├── client.py # UnifiedLLMClient │ │ └── schemas.py # 输入输出数据结构 │ ├── retrieval/ # 检索层 │ │ ├── ingestion.py # 文档解析、切分、入库 │ │ ├── embeddings.py # 向量化服务 │ │ └── search.py # 召回重排 │ ├── agent/ # 编排层 │ │ ├── engine.py # Agent循环 │ │ └── tools.py # 工具注册与调用 │ ├── rag/ # RAG流程组装 │ │ └── pipeline.py # 检索组装上下文模型生成 │ └── evaluation/ # 评估层 │ ├── datasets.py # 评测集管理 │ └── metrics.py # 评分计算 ├── prompts/ # 所有Prompt模板 ├── tests/ # pytest测试 ├── scripts/ # 运维脚本 ├── pyproject.toml # 项目元数据与工具配置 └── requirements-*.txt # 分层依赖3.2 核心代码实现从检索到生成的完整管线下面这段代码是整条管线的枢纽它把检索层、模型层和Prompt模板串起来。我先给出代码再逐行讲解关键设计import asyncio import time from typing import Optional from app.core.config import settings from app.model.client import UnifiedLLMClient from app.model.schemas import ChatMessage from app.retrieval.search import SearchService from app.core.logging import get_logger logger get_logger(__name__) class RAGPipeline: def __init__( self, llm_client: UnifiedLLMClient, search_service: SearchService, ): self.llm llm_client self.search search_service self.top_k settings.RAG_TOP_K # 默认从5开始调优 async def answer( self, question: str, chat_history: Optional[list[ChatMessage]] None, stream: bool False, ) - dict | None: start_time time.perf_counter() # 1. 记录入站请求 logger.info(rag_request_start, questionquestion[:50]) # 2. 召回相关文档 hits await self.search.retrieve(question, top_kself.top_k) if not hits: # 没有召回内容时直接告知用户不进入生成环节 return { answer: 抱歉相关资料库中暂未找到与您问题相关的信息。, sources: [], latency_ms: int((time.perf_counter() - start_time) * 1000), tokens: {prompt: 0, completion: 0}, } # 3. 组装上下文 context_blocks [] for i, hit in enumerate(hits): context_blocks.append(f[{i1}] {hit[content]}) context \n\n.join(context_blocks) # 4. 从Prompt模板目录加载rag_qa规则 prompt_text load_prompt( rag_qa, role智能客服助理, contextcontext, questionquestion, historyformat_history(chat_history) ) # 5. 调用模型生成 messages [{role: user, content: prompt_text}] if stream: # 流式场景把生成器返回给上层 return self._stream_answer(messages, hits, start_time) resp await self.llm.complete(messages) # 6. 结构化日志记录本轮关键指标 logger.info( rag_request_finished, latencyf{time.perf_counter() - start_time:.2f} ) return { answer: resp[content], sources: [hit[metadata] for hit in hits], latency_ms: int((time.perf_counter() - start_time) * 1000), tokens: { prompt: resp[prompt_tokens], completion: resp[completion_tokens], total: resp[prompt_tokens] resp[completion_tokens], }, } async def _stream_answer(self, messages, hits, start_time): 流式响应逐token通过网络发送 async def gen(): content_parts [] async for chunk in self.llm.stream(messages): content_parts.append(chunk) yield chunk # 生成结束后记录完整指标 logger.info( rag_stream_finished, latencytime.perf_counter() - start_time, output_charssum(len(c) for c in content_parts), ) return gen()有几个地方值得划重点**第2步的无召回短路返回**是我在多次实际运营中总结的教训。早期版本在检索为空时仍然调用模型生成模型会一本正经地开始编造——这些内容对用户毫无帮助还白烧token。如果资料库确实没有答案不如直接坦诚告知给用户更可信的交互体验。上下文编号技巧在检索片段前加[1]、[2]这类编号再在Prompt中要求模型引用[1]中的内容回答可以显著提升回答的溯源能力。后续在解析回答时也能通过提取编号来定位回答所引用的源文档片段。3.3 Agent工具调用从问答到行动单轮问答只是AI工程的第一步具备工具调用能力后AI系统才能真正干活。我用一个天气查询工具来演示工具注册和调用的完整链路# app/agent/tools.py import inspect import json from typing import Any, Callable, Coroutine class ToolRegistry: def __init__(self): self._schemas [] self._handlers {} def register(self, handler: Callable[..., Any]) - ToolRegistry: 自动从一个Python函数生成工具Schema避免手写JSON Schema sig inspect.signature(handler) parameters { type: object, properties: {}, required: [] } for name, param in sig.parameters.items(): # 这里简化处理支持str和int类型 schema_type string if param.annotation is str else integer parameters[properties][name] {type: schema_type} if param.default is inspect.Parameter.empty: parameters[required].append(name) self._schemas.append({ type: function, function: { name: handler.__name__, description: inspect.getdoc(handler) or , parameters: parameters } }) self._handlers[handler.__name__] handler return self async def execute(self, name: str, arguments: dict) - str: handler self._handlers.get(name) if not handler: return f错误工具 {name} 不存在 try: # 统一捕获异常防止工具崩溃拖垮整个Agent循环 result await handler(**arguments) return json.dumps(result, ensure_asciiFalse) except Exception as e: return f工具执行失败: {e} # 定义一个天气查询工具 async def get_weather(city: str) - dict: 查询指定城市的天气信息。 参数 city (str): 城市名称如北京 返回 dict: 包含当前温度、天气状况和风力等级 # 实际情况中这里会调用气象服务API return {city: city, temperature: 23, condition: 多云, wind: 二级} registry ToolRegistry() registry.register(get_weather)有了工具注册表之后Agent引擎的核心循环是调用模型 → 模型返回工具调用指令 → 执行工具 → 将结果喂回模型 → 循环直到模型给出最终答案。这个循环看似简单但几个工程细节能决定成败工具执行结果必须截断工具可能返回大量数据超出上下文窗口。我设置了单次工具结果最多3000字符的截断阈值超出部分直接省略并以[已省略N字符]标注避免上下文爆炸。轮数限制必须刚性执行Agent在复杂任务中可能陷入调用工具→继续调用工具的死循环。我设置了最大5轮的限制达到后强制要求模型基于已有信息作答。异步执行不阻塞工具调用凡是涉及IOHTTP请求、数据库查询的一律异步实现保证高并发下的吞吐。3.4 向量库选型与检索接口实现检索环节直接影响RAG生成的忠实度。我实测过多种方案早期用faiss-cpu做本地向量索引数据量小、性能够用后期数据量上到百万级分布式检索和多租户隔离的需求出现迁移到Milvus。给一个选型参考表方案优点缺点适用场景faiss-cpu零运维、轻松集成单机限制、不支持实时动态更新原型验证、千条级数据Chroma轻量级、接口简单性能上限较低、功能较基础中小规模、快速迭代期Qdrant负载均衡好、过滤能力强需要独立部署数据量百万级以下Milvus水平扩展、支持多租户、混合检索部署运维复杂度高百万级数据以上、生产环境无论选哪种我都要在检索服务之上做一层统一的接口抽象这样才不会把自己锁死在某个特定向量库上# app/retrieval/search.py class SearchService: def __init__(self, vector_store, embedder, rerankerNone): self.store vector_store self.embedder embedder self.reranker reranker async def retrieve(self, query: str, top_k: int 5) - list[dict]: # 1. 对查询做向量化 query_vector await self.embedder.embed_query(query) # 2. 向量检索基础召回取更多的候选给重排留余量 candidates self.store.search( vectorquery_vector, top_ktop_k * 3, # 基础召回数量是最终返回的三倍 ) # 3. 重排可选用交叉编码器精排大幅提升相关性 if self.reranker and len(candidates) 1: scored self.reranker.rerank(query, [c[content] for c in candidates]) # scored 是按分数降序的索引列表 candidates [candidates[i] for i in scored[:top_k]] # 4. 包装成统一返回格式 return [ { content: c[content], metadata: c.get(metadata, {}), score: c.get(score, 0.0), } for c in candidates[:top_k] ]关于召回环节的3倍候选重排策略这里多解释一句。向量检索本质上是模糊匹配它确保大概相关的内容被召回来但精排能力有限。交叉编码器重排模型能同时看到查询和文档的完整语义相关性排序精度显著高于单纯向量距离代价是计算成本高、速度慢。所以工程上通行的做法是用向量检索先召回一批候选再用精排模型从候选中挑出最相关的那几个兼顾效率和精度。4. 常见问题与排查技巧实录4.1 模型输出的不稳定问题现象同样的输入模型有时给出完美答案有时答非所问甚至重复输出。排查路径首先确认是否人为原因导致——检查温度参数是否被意外调高如果多个模型版本在负载均衡后切换也会导致行为变化。我遇到过最隐蔽的一次代码里某个模块用默认参数初始化客户端而默认温度恰好不是我们线上约定的0.2导致部分请求抽风。其次要检查Prompt中是否存在歧义。实测发现问题表述里如果包含多个隐含条件模型理解容易漂移。解决方案是把关键约束在Prompt中显式列出。最后对于重复输出的问题可以在解码参数中开启frequency_penalty或者对输出做后处理去重。处理这类问题的核心方法论是先把问题确定下来再找原因。通过结构化的反馈按钮和日志归因快速定位是一类问题还是个例避免在单次偶发问题上浪费过多时间。4.2 检索召回为空或准确率低现象用户提出明确问题时系统却回答资料库中未找到相关信息或者搜出来的资料牛头不对马嘴。原因分析文档切分不合理检索单元过大或过小导致语义不完整查询改写缺失用户口语化表述与文档中的书面语差异大嵌入模型领域适配度差通用模型对特定领域的文本表征效果不佳解决方案对检索命中率做分批测试将召回内容打印出来人工审查而不是只看最终答案质量增加查询改写环节在进入检索前先用一个轻量模型把口语化提问改写成更规范的检索式表达检查索引中是否存在脏数据乱码、水印前缀脏数据会严重污染召回排序4.3 上下文字符串超限现象长对话场景下请求报400错误提示上下文过长。原因分析多轮对话结束后消息列表持续累积最终撑爆上下文窗口。解决方案实现上下文压缩策略。保留第一条系统指令对中间的历史会话做摘要只保留最新几轮原始消息。如果业务允许还可以引入关键信息提取环节——从历史对话中提取用户偏好、未决事项等关键信息注入系统提示词丢弃其余细节。另外我非常推荐在模型层封装中加上发送前预估token的逻辑def estimate_tokens(messages: list[dict]) - int: 粗略估算token数用于预警而非精确计数 total 0 for msg in messages: # 中文约1.5字符/token英文约4字符/token这里用混合估算 total int(len(msg.get(content, )) / 1.8) return total class ContextGuard: def __init__(self, max_context_tokens: int 8000): self.max_context max_context_tokens def trim_messages(self, messages: list[dict]) - list[dict]: while messages and estimate_tokens(messages) self.max_context: # 从最早的非系统消息开始丢弃 for i, msg in enumerate(messages): if msg[role] ! system: messages.pop(i) break return messages4.4 线上性能瓶颈现象用户量上来后接口响应变慢甚至超时。排查步骤先看瓶颈在哪一层模型调用、检索、还是生成环节模型调用耗时长引入流式输出首字延迟大幅下降、考虑量化部署或采用推理加速服务检索环节耗时长检查向量索引是否在内存中、是否命中缓存、是否需要水平扩容引入缓存层对高频问题做精确匹配缓存对语义相似问题做嵌入缓存我给线上服务加的缓存策略分两级精确缓存直接命中完全相同的question时直接返回历史答案近似缓存将query的embedding与最近N条历史查询的embedding做相似度比较相似度超过0.92时复用历史答案class AnswerCache: def __init__(self, redis_client, embedder): self.redis redis_client self.embedder embedder self.sim_threshold 0.92 async def get(self, question: str) - dict | None: # 先检查精确命中 exact await self.redis.get(fexact:{hash(question)}) if exact: return json.loads(exact) # 再检查近似命中 async for key in self.redis.scan_iter(embed:*, count100): cached_embed json.loads(key) sim cosine_similarity(self.embedder.embed_query(question), cached_embed) cached_q key.replace(embed:, ) if sim self.sim_threshold: # 返回已有答案并记录相似度 return json.loads(await self.redis.get(fcache:{hash(cached_q)})) return None这套缓存上线后线上接近30%的重复或相似问题不再触发模型调用既降低了延迟又节省了可观的token成本。5. 经验总结与下一步扩展方向从零构建AI工程这条路走下来我觉得最值的地方不是最终那套代码而是过程中建立的对每一层技术的判断力。现在看到任何一个AI项目我可以快速拆解它的检索策略、上下文管理方式、评估逻辑是否合理这种能力靠看文档是学不来的必须亲手搭过一遍才有体感。有一个体会想分享给正在做类似项目的朋友不要试图在第一版就追求完美。AI工程本身是一门实验性很强的工程学科很多设计决策比如切分粒度、TopK数值、温度参数并没有理论的唯一最优解而是依赖你的数据分布和目标场景。我的做法是把默认参数设置为一个合理值然后靠评估体系去驱动调优而不是在搭建阶段就陷入参数泥潭。关注几个值得扩展的方向多模态能力接入在现有文本管线基础上扩展图片输入和表格理解能力这需要增加视觉模型的统一封装层流式工作流与人工审核高风险场景如医疗建议、法律咨询需要在AI生成后加入人工审核节点编排层需要支持暂停和恢复机制LLM应用的可观测性当前用结构化日志已经能覆盖大部分排障需求但更复杂的链路追踪尤其是多Agent协作场景可以接入OpenTelemetry体系另外一个实际心得保持依赖的轻量。我在项目中尽量只依赖那些低层的工具库——FastAPI、Jinja2、pydantic、Redis而对那些高层的AI编排框架保持审慎。框架解决问题也框定思路。当你对每一层都有掌控力之后再回头去看那些框架你会更清楚它们在替你做什么、以及它们的边界在哪。