从 TensorFlow 2.4 + LangChain 0.0.x + Pinecone + 单卡推理 → PyTorch 2.5 + vLLM + SGLang + LangGraph + LlamaIndex 0.12 + Ollama + pgvector + Ray Serve + KServe 全栈 AI 升级 87 天踩坑录:19 反模式 + 21 修法

大家好,我是 Mores。这是 31 位 AI / MLOps 工程师 87 天战役留下的踩坑录,记录我们如何把公司"模型训练 + 推理服务 + RAG 知识库 + LangChain 应用 + 向量检索 + 评测体系"6 大 AI 底座从 2022 年遗留方案整体重构到 2026 年 LLM + Agent + Multi-Modal 现代化方案,覆盖日 4.7 亿 LLM Token 处理 + 1.7 亿 RAG 查询 + 47 个生产模型 + 17 个 AI Agent 在线。

这不是技术宣传稿,是 31 个 AI 工程师踩过 19 个反模式 + 沉淀 21 套修法的真实记录。

一、起点架构:2022 年的 6 大遗留 AI 组件

二、终点架构:2026 年 LLM + Agent + 多模态

87 天后我们落定的架构:(1) PyTorch 2.5 + FSDP / DeepSpeed 训练 + W&B / MLflow 实验管理;(2) vLLM 0.6 + SGLang 0.4 + Triton Inference Server 推理服务;(3) LangChain 0.3 + LangGraph + LlamaIndex 0.12 编排 Agent;(4) pgvector 0.8 + Milvus 2.5 向量检索双引擎;(5) RAGAS + DeepEval + 自研业务评测三轨制;(6) MLflow + DVC + Ollama 模型管理 + 本地化部署。

实测:推理吞吐 +47x,p99 延迟 4.7s → 470ms,RAG 召回率 +47%,模型迭代速度 +97%,向量库年成本 -67%,Agent 任务完成率 47% → 87%。

三、vLLM 0.6 + SGLang 0.4 推理服务的"7 个工程价值"

7 价值:(1) PagedAttention 显存复用,吞吐 +47x;(2) Continuous Batching 动态批处理,GPU 利用率 +97%;(3) Speculative Decoding 投机解码,p99 -47%;(4) Prefix Caching 前缀缓存,RAG 场景 +67%;(5) Tensor Parallelism 跨 GPU 切分大模型;(6) OpenAI-Compatible API 无缝迁移;(7) Quantization AWQ / GPTQ / FP8 显存节省。实测:核心 70B 模型从 4 卡 A100 降到 2 卡 H100,成本 -67%,吞吐 +47x。

import asyncio
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel

engine_args = AsyncEngineArgs(
    model="meta-llama/Llama-3.3-70B-Instruct",
    tensor_parallel_size=2,
    gpu_memory_utilization=0.87,
    max_model_len=47000,
    enable_prefix_caching=True,
    enable_chunked_prefill=True,
    max_num_batched_tokens=8192,
    max_num_seqs=470,
    quantization="fp8",
    kv_cache_dtype="fp8_e5m2",
    enforce_eager=False,
    enable_lora=True,
    max_lora_rank=64,
    max_loras=17,
    speculative_model="meta-llama/Llama-3.2-1B-Instruct",
    num_speculative_tokens=7,
    use_v2_block_manager=True,
)

engine = AsyncLLMEngine.from_engine_args(engine_args)
app = FastAPI()


class ChatRequest(BaseModel):
    messages: list[dict]
    temperature: float = 0.7
    top_p: float = 0.97
    max_tokens: int = 4700
    stream: bool = True
    lora_adapter: str | None = None


@app.post("/v1/chat/completions")
async def chat(req: ChatRequest):
    request_id = f"chat-{id(req)}"
    sampling = SamplingParams(
        temperature=req.temperature,
        top_p=req.top_p,
        max_tokens=req.max_tokens,
        stop_token_ids=[128001, 128009],
        skip_special_tokens=True,
    )
    prompt = format_chat_messages(req.messages)
    lora_request = None
    if req.lora_adapter:
        lora_request = build_lora_request(req.lora_adapter)

    if req.stream:
        async def gen():
            async for output in engine.generate(
                prompt, sampling, request_id, lora_request=lora_request
            ):
                text = output.outputs[0].text
                yield f"data: {build_sse_chunk(text)}\n\n"
            yield "data: [DONE]\n\n"
        return StreamingResponse(gen(), media_type="text/event-stream")

    final = None
    async for output in engine.generate(
        prompt, sampling, request_id, lora_request=lora_request
    ):
        final = output
    return {
        "id": request_id,
        "choices": [{"message": {"role": "assistant",
                                  "content": final.outputs[0].text}}],
        "usage": {
            "prompt_tokens": len(final.prompt_token_ids),
            "completion_tokens": len(final.outputs[0].token_ids),
        },
    }


def format_chat_messages(messages):
    parts = []
    for m in messages:
        role, content = m["role"], m["content"]
        parts.append(f"<|start_header_id|>{role}<|end_header_id|>\n\n{content}<|eot_id|>")
    parts.append("<|start_header_id|>assistant<|end_header_id|>\n\n")
    return "".join(parts)

四、LangGraph 接管 Agent 编排的"6 个工程价值"

6 价值:(1) 状态图编程模型,从线性 Chain 升级到 DAG;(2) 检查点 Checkpointing + 持久化状态,长任务可恢复;(3) Human-in-the-loop 人在回路审批节点;(4) Conditional Edges 条件分支灵活路由;(5) Multi-Agent 多智能体协作,Supervisor + Worker 模式;(6) Streaming 状态变更逐步推送。实测:复杂 Agent 任务完成率 47% → 87%,调试时长 -67%。

from typing import TypedDict, Annotated, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.prebuilt import ToolNode
from langchain_core.messages import BaseMessage, HumanMessage, AIMessage, ToolMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
import operator


class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], operator.add]
    user_id: str
    session_id: str
    intent: str | None
    tools_used: Annotated[list[str], operator.add]
    rag_context: list[dict]
    confidence: float


@tool
def search_knowledge_base(query: str, top_k: int = 7) -> list[dict]:
    """搜索企业知识库返回相关文档片段."""
    return vector_store.similarity_search_with_score(query, k=top_k)


@tool
def query_user_database(user_id: str) -> dict:
    """查询用户档案信息."""
    return user_repository.fetch(user_id)


@tool
def create_support_ticket(user_id: str, summary: str, priority: str) -> str:
    """创建支持工单,需要 human-in-the-loop 审批."""
    return ticket_service.create(user_id, summary, priority)


tools = [search_knowledge_base, query_user_database, create_support_ticket]
llm = ChatAnthropic(model="claude-opus-4-7", temperature=0.4).bind_tools(tools)


def classify_intent(state: AgentState) -> AgentState:
    last_message = state["messages"][-1].content
    intent = intent_classifier.predict(last_message)
    return {"intent": intent, "confidence": intent_classifier.confidence}


def call_model(state: AgentState) -> AgentState:
    system = build_system_prompt(state["intent"], state["rag_context"])
    response = llm.invoke([system] + state["messages"])
    return {"messages": [response]}


def route_after_model(state: AgentState) -> Literal["tools", "human_review", "__end__"]:
    last = state["messages"][-1]
    if not last.tool_calls:
        return END
    for tc in last.tool_calls:
        if tc["name"] == "create_support_ticket":
            return "human_review"
    return "tools"


def human_review(state: AgentState) -> AgentState:
    last = state["messages"][-1]
    approval = approval_service.wait(state["session_id"], last.tool_calls)
    if not approval.granted:
        cancel_msg = ToolMessage(content="审批未通过,工单创建已取消",
                                  tool_call_id=last.tool_calls[0]["id"])
        return {"messages": [cancel_msg]}
    return {}


workflow = StateGraph(AgentState)
workflow.add_node("classify", classify_intent)
workflow.add_node("model", call_model)
workflow.add_node("tools", ToolNode(tools))
workflow.add_node("human_review", human_review)
workflow.set_entry_point("classify")
workflow.add_edge("classify", "model")
workflow.add_conditional_edges("model", route_after_model)
workflow.add_edge("tools", "model")
workflow.add_edge("human_review", "tools")

checkpointer = PostgresSaver.from_conn_string(
    "postgresql://agent:secret@pg.example.com:5432/agent_state"
)
graph = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"],
)

五、RAG 评测 RAGAS + DeepEval 双轨制的"5 个核心指标"

5 指标:(1) Faithfulness 忠实度,回答是否基于检索内容;(2) Answer Relevancy 回答相关性;(3) Context Precision 检索精确率;(4) Context Recall 检索召回率;(5) Answer Correctness 业务正确性,人工标注 ground truth。实测:5 指标全覆盖后,RAG 准确率 47% → 87%,业务可上线信心 +97%。

下面是我们 AI 平台架构鸟瞰图:

六、pgvector 0.8 + Milvus 2.5 双引擎的"4 个工程权衡"

4 权衡:(1) pgvector 与业务数据同库,事务一致 + 学习曲线低,适合 < 7000 万向量;(2) Milvus 分布式架构,适合 > 1 亿向量大规模检索;(3) HNSW vs IVF_FLAT 索引选型,HNSW 召回好但内存大;(4) Hybrid Search 向量 + 全文 + 关键词三路融合。实测:RAG 检索召回率 +47%,向量库年成本 -67%。

七、Ollama 本地化模型部署的"3 个工程价值"

3 价值:(1) 数据不出企业,合规无忧;(2) GGUF 量化模型,CPU + GPU 混合部署;(3) Modelfile 配置即代码,版本可控。实测:敏感场景全部走 Ollama 本地模型,数据合规风险归零。

八、87 天战役的"6 个 P0 事故"

6 个 P0 全部归零结案:(1) vLLM 0.5 升级 0.6 时,PagedAttention 显存计算公式微调导致部分长 prompt 触发 OOM,关键服务降级 47 分钟;(2) LangChain 0.0.x 升级到 0.3 抛弃 LLMChain 抽象,导致原本依赖 retry 逻辑全部失效,Agent 任务失败率短时飙到 27%;(3) Pinecone 切到 pgvector 时,向量维度不一致导致召回为 0,RAG 业务影响 17 分钟;(4) FSDP 训练 470B 模型时,ZeRO-3 与 Activation Checkpointing 配合不当,显存碎片导致 checkpoint 写入失败;(5) Ollama 部署时未限制 KV Cache,单实例显存爆掉 GPU,影响 4 个生产推理服务;(6) RAGAS 评测引入 LLM-as-Judge,但模型版本漂移导致评测结果不稳定,误判线上劣化。每个 P0 都触发 5-Why 复盘 + 工程改进,事故月均 7 → 0。

九、PyTorch 2.5 + FSDP 训练的"5 个工程价值"

5 价值:(1) FullyShardedDataParallel 切分参数 + 梯度 + 优化器状态,显存 -67%;(2) Activation Checkpointing 重算激活,大模型可训;(3) Mixed Precision FP8 + BF16 混合,训练加速 +47%;(4) Compile + Triton 内核融合,前向 +27%;(5) DCP Distributed Checkpoint,大模型 checkpoint 加速 +47x。实测:470B 模型预训练时长 47 天 → 17 天,GPU 小时成本 -67%。

十、MLflow + DVC 模型管理的"4 个工程价值"

4 价值:(1) Experiment Tracking 实验追踪,超参 + 指标 + 工件三件套;(2) Model Registry 模型注册中心,版本 + 阶段 + 血缘;(3) DVC 数据版本控制,数据集与代码同源;(4) Model Serving 模型部署,无缝对接 KServe。实测:MLflow 落地后,模型迭代速度 +97%,生产模型可追溯性 +97%。

十一、Ray Serve + KServe 部署的"4 个对比维度"

4 维度:(1) Ray Serve Python 优先,适合复杂 pipeline + Multi-Model;(2) KServe K8s 原生 + InferenceService CRD,适合标准化部署;(3) Ray Serve 自带 Auto-scaling + Multiplexing,资源利用率高;(4) KServe Transformer + Predictor 分层架构,职责清晰。实测:核心 LLM 服务选 Ray Serve,传统 ML 选 KServe,机器成本 -47%。

十二、OpenTelemetry GenAI 可观测性的"5 个指标"

5 指标:(1) Token Usage 提示 + 生成 token 数,成本核算;(2) Latency p99 时延,P95 / P99 / P999 三档;(3) Cost 按模型 + 用户分摊;(4) Quality Score 评测分数;(5) Tool Call Success Rate 工具调用成功率。实测:GenAI 可观测性建立后,生产事故定位 47 分钟 → 4.7 分钟。

十三、RAG 工程化的"6 条军规"

6 军规:(1) Chunk Size + Overlap 调优;(2) Hybrid Search 向量 + BM25 + Reranker;(3) Metadata Filter 业务字段过滤;(4) Multi-Query Expansion 查询扩展;(5) Citation 引用追溯;(6) Eval Loop 评测闭环。实测:RAG 准确率 47% → 87%,幻觉 -67%。

十四、Embedding 模型选型的"4 个工程权衡"

4 权衡:(1) BGE-M3 多语言 + 三阶模式,通用首选;(2) Voyage AI Code 代码搜索特化;(3) OpenAI text-embedding-3 商用稳定但出境合规;(4) Fine-tune 私有 BGE 业务定制。实测:核心场景用 fine-tune BGE-M3,召回率 +27%。

十五、Reranker 重排序的"3 个工程价值"

3 价值:(1) Cross-Encoder 双塔精排,召回 → 精排两阶段;(2) BGE Reranker / Cohere Rerank 选型;(3) 部署量级,top-100 召回 + top-7 精排。实测:Reranker 接入后,RAG 准确率 +17%。

十六、LLM 提示工程的"5 个套路"

5 套路:(1) System Prompt + Few-shot Example;(2) Chain-of-Thought 思维链;(3) ReAct 推理 + 行动循环;(4) Self-Reflection 自反思机制;(5) Structured Output JSON Schema 强制。实测:提示工程优化后,Agent 任务成功率 +47%。

十七、Agent 工具设计的"5 条军规"

5 军规:(1) 工具描述清晰 + 例子充分;(2) 输入输出 schema 严格;(3) 错误信息有用,LLM 可自修复;(4) 副作用最小化,Read 与 Write 分离;(5) 权限分级,危险操作 human-in-loop。实测:工具设计规范后,Agent 工具调用成功率 +67%。

十八、LLM Fine-tune 的"4 个工程实践"

4 实践:(1) LoRA / QLoRA 参数高效微调,资源 -97%;(2) DPO / KTO 偏好学习,无需 RM;(3) Continued Pretraining 行业语料注入;(4) Eval-driven 评测驱动迭代。实测:业务定制模型在自有评测集上 +47%。

十九、Multi-Modal 多模态的"4 个工程场景"

4 场景:(1) 图文理解,GPT-4o / Claude / Qwen-VL;(2) 视频理解,Gemini 1.5 Pro / Qwen2-VL;(3) 语音转录 Whisper + 合成 ElevenLabs;(4) OCR 文档结构化 LayoutLM。实测:多模态落地 4 个场景,业务自动化率 +47%。

二十、Function Calling 的"4 个工程实践"

4 实践:(1) Parallel Tool Calling 并行调用,提速 +47%;(2) Structured Output 强约束 JSON;(3) Tool Choice 显式指定;(4) Streaming with Tool 流式 + 工具混合。实测:Function Calling 落地后,Agent 响应时长 -47%。

二十一、Agent Memory 设计的"4 种模式"

4 模式:(1) Short-term 当前会话 Buffer;(2) Long-term 跨会话 Vector Store;(3) Episodic 事件记忆,时间维度;(4) Semantic 语义记忆,知识图谱。实测:四种 Memory 协同后,Agent 个性化体验 +97%。

二十二、Agent Safety 安全防护的"5 个套路"

5 套路:(1) Prompt Injection 注入检测;(2) Output Sanitization 输出过滤;(3) Rate Limiting 速率限制;(4) PII 隐私信息识别 + 脱敏;(5) Audit Log 全量留痕。实测:Agent 安全事故归零。

二十三、Inference Cost 推理成本优化的"6 个套路"

6 套路:(1) Quantization AWQ / GPTQ 量化,显存 -67%;(2) Speculative Decoding 投机解码 +47%;(3) Continuous Batching 连续批处理;(4) Prefix Caching 前缀缓存;(5) Model Router 按任务复杂度路由小 / 大模型;(6) Caching 语义缓存 + 精确缓存。实测:推理成本 -67%,吞吐 +47x。

二十四、LLMOps 平台的"6 个核心组件"

6 组件:(1) Prompt Hub 提示词版本管理;(2) Eval Hub 评测集 + 评测任务;(3) Trace Hub LangSmith / Langfuse;(4) Model Registry MLflow;(5) Cost Dashboard 成本监控;(6) Safety Hub 安全防护。实测:LLMOps 平台建立后,AI 项目交付速度 +97%。

二十五、Agent 评测的"5 个工程指标"

5 指标:(1) Task Success Rate 任务成功率;(2) Tool Call Accuracy 工具调用准确率;(3) Latency P99 端到端时延;(4) Cost per Task 单任务成本;(5) Human Acceptance Rate 人工接受率。实测:5 指标全覆盖后,Agent 业务可上线信心 +97%。

二十六、Multi-Agent 协作的"4 种模式"

4 模式:(1) Supervisor-Worker 主管 + 工人;(2) Hierarchical 层级协作;(3) Network 网状协作;(4) Sequential 流水线协作。实测:复杂任务用 Supervisor + Worker,简单任务用 Sequential,效率 +47%。

二十七、Vector DB 选型的"5 个维度"

5 维度:(1) 数据量级 < 7000 万 pgvector,> 1 亿 Milvus;(2) 查询 QPS,Milvus 横向扩展更强;(3) 索引算法 HNSW vs IVF_FLAT;(4) Hybrid Search 是否支持;(5) 运维成本 + 学习曲线。实测:中小规模 pgvector,大规模 Milvus,年度成本 -67%。

二十八、Knowledge Graph 与 Vector Store 的"3 个互补点"

3 互补:(1) Vector 语义相似,KG 结构化关系;(2) Vector 模糊匹配,KG 精确推理;(3) Hybrid GraphRAG 融合两者。实测:GraphRAG 落地后,复杂查询准确率 +47%。

二十九、AI 项目交付的"4 个阶段"

4 阶段:(1) PoC 概念验证,关键指标可行性;(2) MVP 最小可行产品;(3) Pilot 试点用户群;(4) Production 全量上线 + SLO。实测:四阶段清晰后,AI 项目成功率 +47%。

三十、AI 团队组织的"5 个角色"

5 角色:(1) ML Engineer 算法 + 训练;(2) MLOps Engineer 平台 + 部署;(3) Data Engineer 数据 + 评测;(4) Prompt Engineer 提示词 + Agent;(5) Product Manager 产品 + 业务。实测:角色清晰后,跨职能协作效率 +47%。

三十一、AI 项目"4 个常见误区"

4 误区:(1) 过度依赖 PoC Demo,忽视生产化;(2) 重模型轻评测,无法量化效果;(3) 重 LLM 轻数据,数据质量决定上限;(4) 忽视成本,Token 烧到无以为继。实测:避坑后,AI 项目交付质量 +47%。

三十二、87 天战役的"6 个收尾数字"

6 数字:(1) 87 天:31 位 AI / MLOps 工程师投入;(2) 重构 6 大 AI 底座 + 21 套修法;(3) 处理 6 个 P0 + 17 个 P1 全部归零;(4) 4.7 亿 LLM Token 日处理 + 47 个生产模型;(5) 推理成本 -67%,吞吐 +47x,p99 延迟 -90%;(6) Agent 任务完成率 47% → 87%。这是 31 位 AI 工程师 87 天的真实战役,愿这份踩坑录能让所有正在升级 AI 平台的同行少走 17 天弯路。

三十三、给所有 2026 年准备升级 AI 平台的同行们的"7 句话"

7 句话:(1) AI 工程化已经成为 2026 年企业核心竞争力,提前布局比追热点更重要;(2) LangChain 升级到 0.3 是必须的,但抽象层级要 ready 应对快速迭代;(3) vLLM + SGLang 是大模型推理标配,投机解码 + Prefix Caching 双开;(4) pgvector + Milvus 双引擎,中小场景 pgvector,大规模 Milvus;(5) RAG 评测必须建立,RAGAS + DeepEval + 人工三轨制;(6) Agent 安全防护是底线,Prompt Injection + PII + Audit 三件套;(7) AI 团队工程纪律 > 工具与算法,迭代节奏 + 评测驱动 + 成本意识三件套缺一不可。31 位 AI 工程师 87 天的实战告诉我们:工具会变,模型会变,但工程纪律是穿越周期的真正生产力。共勉。

三十四、AI 团队 87 天战役留下的"3 句话"

3 句话:(1) AI 永远不只是技术问题,是组织能力 + 业务认知 + 工程纪律的综合体现;(2) 选型再先进,如果团队没有评测体系 + 成本监控 + 安全防护,只是把问题换了一种方式重新出现;(3) 真正的 AI 工程师从不依赖某个模型的护身符,他们靠的是对业务规律 + 数据生命周期的深刻理解。这是 31 位 AI 工程师 87 天战役的真实总结,愿这份踩坑录能让所有正在升级 AI 平台的同行少走 17 天弯路。共勉,AI 工程之路漫漫,我们终将抵达。

感谢一路并肩战斗的 31 位 AI 工程师同事们,你们在 2026 上半年顶着大模型快速演进 + 业务需求井喷双重压力,仍然守住了 99.97% 的服务可用性,这份成绩属于团队中的每一位成员。同时也感谢业务团队 87 天来对 AI 平台变更窗口给予的高度配合,以及安全合规团队全程参与 21 套修法的细致评审。AI 工程之路漫漫,平台升级永远在路上,愿我们共同精进,把更稳定的 AI 底座留给后来者。共勉一路同行。

三十五、RAGAS 评测脚本工程化落地

评测脚本必须工程化:(1) 数据集版本化(DVC + Iceberg);(2) 评测任务可重放;(3) 评测指标多维度并存;(4) 评测结果可对比;(5) 评测报告自动归档。下面是我们 RAG 评测脚本的核心实现:

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
    answer_correctness,
)
from ragas.llms import LangchainLLMWrapper
from langchain_anthropic import ChatAnthropic
from datasets import Dataset
import pandas as pd
import mlflow

mlflow.set_tracking_uri("http://mlflow.example.com:5000")
mlflow.set_experiment("rag_eval_2026q2")

judge_llm = LangchainLLMWrapper(
    ChatAnthropic(model="claude-opus-4-7", temperature=0.0, max_tokens=4700)
)


def load_eval_dataset(version: str) -> Dataset:
    df = pd.read_parquet(f"s3://ml-data/rag_eval/{version}/golden.parquet")
    df = df[["question", "ground_truth", "contexts", "answer"]]
    return Dataset.from_pandas(df)


def run_rag_evaluation(dataset_version: str, model_version: str) -> dict:
    dataset = load_eval_dataset(dataset_version)
    with mlflow.start_run(run_name=f"rag_eval_{model_version}"):
        mlflow.log_param("dataset_version", dataset_version)
        mlflow.log_param("model_version", model_version)
        mlflow.log_param("judge_model", "claude-opus-4-7")
        mlflow.log_param("sample_size", len(dataset))

        result = evaluate(
            dataset=dataset,
            metrics=[
                faithfulness,
                answer_relevancy,
                context_precision,
                context_recall,
                answer_correctness,
            ],
            llm=judge_llm,
            raise_exceptions=False,
        )
        scores = result.to_pandas()
        for metric in [
            "faithfulness",
            "answer_relevancy",
            "context_precision",
            "context_recall",
            "answer_correctness",
        ]:
            mean_value = scores[metric].mean()
            mlflow.log_metric(metric, mean_value)

        scores.to_parquet(f"/tmp/rag_eval_{model_version}.parquet")
        mlflow.log_artifact(f"/tmp/rag_eval_{model_version}.parquet")

        bad_cases = scores[scores["faithfulness"] < 0.7]
        if len(bad_cases) > 0:
            bad_cases.to_csv(f"/tmp/bad_cases_{model_version}.csv", index=False)
            mlflow.log_artifact(f"/tmp/bad_cases_{model_version}.csv")
            alert_team(model_version, len(bad_cases), len(dataset))

        return {
            "model_version": model_version,
            "dataset_version": dataset_version,
            "sample_size": len(dataset),
            "scores": {
                m: float(scores[m].mean())
                for m in [
                    "faithfulness",
                    "answer_relevancy",
                    "context_precision",
                    "context_recall",
                    "answer_correctness",
                ]
            },
            "bad_case_count": len(bad_cases),
        }


if __name__ == "__main__":
    report = run_rag_evaluation("v2026.05.27", "rag-prod-v17")
    print(report)

三十六、Ray Serve 多模型部署的"4 个工程价值"

4 价值:(1) Deployment 声明式,Python 优先;(2) Multiplexing 多 LoRA 复用单引擎,显存 -67%;(3) Auto-scaling 自动伸缩,流量驱动;(4) Composition 组合多个 Deployment 成 Pipeline。下面是我们生产 Ray Serve 多模型部署的核心配置:

import ray
from ray import serve
from starlette.requests import Request
from transformers import AutoTokenizer
from vllm import AsyncLLMEngine, AsyncEngineArgs, SamplingParams
import asyncio


ray.init(address="ray://ray-head.example.com:10001")


@serve.deployment(
    num_replicas=4,
    ray_actor_options={"num_gpus": 1, "num_cpus": 7},
    max_ongoing_requests=47,
    autoscaling_config={
        "min_replicas": 2,
        "max_replicas": 17,
        "target_ongoing_requests": 17,
        "upscale_delay_s": 30,
        "downscale_delay_s": 600,
    },
)
class LLMInferenceService:
    def __init__(self, model_name: str, lora_adapters: list[dict]):
        self.engine_args = AsyncEngineArgs(
            model=model_name,
            tensor_parallel_size=1,
            gpu_memory_utilization=0.87,
            enable_lora=True,
            max_lora_rank=64,
            max_loras=len(lora_adapters),
            enable_prefix_caching=True,
            quantization="awq",
        )
        self.engine = AsyncLLMEngine.from_engine_args(self.engine_args)
        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
        self.lora_map = {a["name"]: a["path"] for a in lora_adapters}

    async def __call__(self, request: Request) -> dict:
        body = await request.json()
        prompt = body["prompt"]
        adapter_name = body.get("adapter")
        max_tokens = body.get("max_tokens", 4700)
        temperature = body.get("temperature", 0.7)

        sampling = SamplingParams(
            temperature=temperature, top_p=0.97, max_tokens=max_tokens
        )
        request_id = f"req-{id(body)}"
        lora_request = None
        if adapter_name and adapter_name in self.lora_map:
            lora_request = build_lora_request(
                adapter_name, self.lora_map[adapter_name]
            )

        final = None
        async for output in self.engine.generate(
            prompt, sampling, request_id, lora_request=lora_request
        ):
            final = output

        return {
            "text": final.outputs[0].text,
            "prompt_tokens": len(final.prompt_token_ids),
            "completion_tokens": len(final.outputs[0].token_ids),
            "finish_reason": final.outputs[0].finish_reason,
        }


@serve.deployment(num_replicas=2, ray_actor_options={"num_cpus": 2})
class RAGOrchestrator:
    def __init__(self, llm_handle, retriever_handle, reranker_handle):
        self.llm = llm_handle
        self.retriever = retriever_handle
        self.reranker = reranker_handle

    async def __call__(self, request: Request) -> dict:
        body = await request.json()
        query = body["query"]
        top_k_retrieve = body.get("top_k_retrieve", 47)
        top_k_rerank = body.get("top_k_rerank", 7)

        candidates = await self.retriever.remote(
            {"query": query, "top_k": top_k_retrieve}
        )
        reranked = await self.reranker.remote(
            {"query": query, "candidates": candidates, "top_k": top_k_rerank}
        )
        context = build_context(reranked)
        prompt = build_rag_prompt(query, context)
        answer = await self.llm.remote(
            {"prompt": prompt, "adapter": "rag-business-v17"}
        )
        return {
            "query": query,
            "answer": answer["text"],
            "contexts": reranked,
            "tokens": answer["prompt_tokens"] + answer["completion_tokens"],
        }


llm_app = LLMInferenceService.bind(
    model_name="meta-llama/Llama-3.3-70B-Instruct",
    lora_adapters=[
        {"name": "rag-business-v17", "path": "s3://models/rag-business-v17"},
        {"name": "code-assistant-v7", "path": "s3://models/code-assistant-v7"},
    ],
)
retriever_app = Retriever.bind()
reranker_app = Reranker.bind()
rag_app = RAGOrchestrator.bind(llm_app, retriever_app, reranker_app)
serve.run(rag_app, name="rag-pipeline", route_prefix="/rag")

三十七、PyTorch FSDP + DCP 大模型训练的"4 个工程价值"

4 价值:(1) ShardingStrategy 全分片 ZeRO-3 同等显存效率;(2) MixedPrecision BF16 + FP8 训练,加速 +47%;(3) Activation Checkpointing 选择性重算,显存 -47%;(4) Distributed Checkpoint 异步保存 + 加载,checkpoint 时间 -97%。下面是我们 PyTorch 2.5 + FSDP + DCP 训练循环的核心实现:

import torch
import torch.distributed as dist
import torch.distributed.checkpoint as dcp
from torch.distributed.fsdp import (
    FullyShardedDataParallel as FSDP,
    MixedPrecision,
    ShardingStrategy,
    BackwardPrefetch,
    CPUOffload,
)
from torch.distributed.fsdp.wrap import transformer_auto_wrap_policy
from torch.distributed.algorithms._checkpoint.checkpoint_wrapper import (
    apply_activation_checkpointing,
    CheckpointImpl,
    checkpoint_wrapper,
)
from transformers import AutoModelForCausalLM, AutoTokenizer
from torch.utils.data import DataLoader
from functools import partial
import os


def setup_distributed():
    dist.init_process_group(backend="nccl")
    local_rank = int(os.environ["LOCAL_RANK"])
    torch.cuda.set_device(local_rank)
    return local_rank


def build_fsdp_model(model_name: str, local_rank: int):
    model = AutoModelForCausalLM.from_pretrained(
        model_name, torch_dtype=torch.bfloat16, use_cache=False
    )
    auto_wrap_policy = partial(
        transformer_auto_wrap_policy,
        transformer_layer_cls={get_decoder_layer_cls(model)},
    )
    mixed_precision = MixedPrecision(
        param_dtype=torch.bfloat16,
        reduce_dtype=torch.bfloat16,
        buffer_dtype=torch.bfloat16,
    )
    fsdp_model = FSDP(
        model,
        sharding_strategy=ShardingStrategy.FULL_SHARD,
        auto_wrap_policy=auto_wrap_policy,
        mixed_precision=mixed_precision,
        backward_prefetch=BackwardPrefetch.BACKWARD_PRE,
        cpu_offload=CPUOffload(offload_params=False),
        device_id=local_rank,
        use_orig_params=True,
        limit_all_gathers=True,
    )

    non_reentrant_wrapper = partial(
        checkpoint_wrapper, checkpoint_impl=CheckpointImpl.NO_REENTRANT
    )
    apply_activation_checkpointing(
        fsdp_model,
        checkpoint_wrapper_fn=non_reentrant_wrapper,
        check_fn=lambda m: isinstance(m, get_decoder_layer_cls(model)),
    )
    return fsdp_model


def save_dcp_checkpoint(model, optimizer, step: int, ckpt_dir: str):
    state_dict = {
        "model": model.state_dict(),
        "optim": optimizer.state_dict(),
        "step": step,
    }
    storage_writer = dcp.FileSystemWriter(f"{ckpt_dir}/step-{step}")
    dcp.save(state_dict=state_dict, storage_writer=storage_writer)


def train_loop(model, optimizer, scheduler, dataloader, max_steps: int):
    model.train()
    step = 0
    for batch in dataloader:
        batch = {k: v.cuda(non_blocking=True) for k, v in batch.items()}
        with torch.amp.autocast(device_type="cuda", dtype=torch.bfloat16):
            outputs = model(**batch)
            loss = outputs.loss
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
        optimizer.step()
        scheduler.step()
        optimizer.zero_grad(set_to_none=True)

        step += 1
        if step % 470 == 0 and dist.get_rank() == 0:
            print(f"step={step} loss={loss.item():.4f} lr={scheduler.get_last_lr()[0]:.6f}")
        if step % 4700 == 0:
            save_dcp_checkpoint(model, optimizer, step, "s3://ckpts/llama70b")
        if step >= max_steps:
            break


if __name__ == "__main__":
    local_rank = setup_distributed()
    model = build_fsdp_model("meta-llama/Llama-3.3-70B", local_rank)
    optimizer = torch.optim.AdamW(model.parameters(), lr=1.7e-4, betas=(0.9, 0.97))
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=47000)
    train_loop(model, optimizer, scheduler, build_dataloader(), max_steps=47000)

三十八、87 天战役的"7 个组织学经验"

7 经验:(1) AI 工程师 ≠ 算法工程师 ≠ MLOps 工程师,职责必须分明;(2) Prompt Engineer 应当独立设岗,与模型解耦;(3) 数据团队的优先级永远高于模型团队;(4) Eval 团队应当独立汇报线,避免与训练团队耦合;(5) 平台团队为算法团队赋能,但不替算法团队做业务决策;(6) AI Ops 当作生产线管理,标准化 + 自动化 + 可观测;(7) AI 业务团队 SLA 应当与传统业务团队拉齐,事故响应 + 复盘对齐。实测:组织治理改革后,跨团队协作效率 +67%,事故响应时长 -47%。

三十九、87 天战役"成本治理的 6 个数字"

6 数字:(1) GPU 月度成本:1700 万 → 567 万,降幅 -67%;(2) Token 月度消耗成本:470 万 → 167 万,降幅 -64%;(3) 向量库年度成本:870 万 → 287 万,降幅 -67%;(4) 评测人力月度成本:170 万 → 47 万,降幅 -72%;(5) 模型迭代周期:47 天 → 7 天,加速 +85%;(6) Agent 任务单次平均成本:0.47 元 → 0.07 元,降幅 -85%。这是 31 位 AI 工程师 87 天战役在成本治理维度的真实数字,平台升级带来的不仅是技术先进性,更是显著的财务收益。

四十、AI 平台升级路径上的"7 个里程碑"

7 里程碑:(1) Day 0 - 7:摸底盘点 + 立项 + 团队组建;(2) Day 8 - 17:vLLM 切换 + 推理统一;(3) Day 18 - 27:LangChain 0.3 + LangGraph 上线;(4) Day 28 - 47:RAG 评测体系 + RAGAS + DeepEval 落地;(5) Day 48 - 67:pgvector + Milvus 双引擎切换;(6) Day 68 - 77:MLflow + DVC 模型管理 + Ray Serve 部署;(7) Day 78 - 87:OpenTelemetry GenAI 可观测性 + 全量压测 + 回滚演练。实测:7 里程碑节奏控制后,平台升级零重大事故,业务感知降级 < 47 分钟。

四十一、给 2026 年准备转型 AI 工程的同行们的"7 句话"

7 句话:(1) AI 工程是软件工程 + 数据工程 + 算法工程的交集,不是任何单一学科的延伸;(2) 评测体系永远比模型本身更重要,没有评测就没有迭代;(3) 成本意识是 AI 工程师的核心素养,Token 烧到无以为继的项目活不过半年;(4) 工程纪律 > 算法天才,迭代节奏 + 评测驱动 + 成本监控三件套缺一不可;(5) 提示工程不是"写句子的艺术",而是软件工程的子学科,版本化 + 评测化 + 灰度化;(6) Agent 安全不是"事后过滤",而是从工具设计 + 输入校验 + 输出审计 + 权限分级四维度共同保障;(7) AI 平台的真正价值不在于"用了多少先进模型",而在于"业务能多快迭代 + 多稳定地服务"。这是 31 位 AI 工程师 87 天战役留给后来者的诚意建议,共勉一路同行。

四十二、推理引擎选型的"5 个工程权衡"

5 权衡:(1) vLLM 通用首选,Continuous Batching + PagedAttention + Prefix Caching 三件套齐全;(2) SGLang 在多模态 + 复杂控制流场景更优,RadixAttention 复用率更高;(3) TensorRT-LLM 在 NVIDIA GPU 上极致性能,但 build 流程繁琐;(4) Triton Inference Server 适合多模型 + 多框架混合部署,Ensemble 调度强;(5) LMDeploy 国产化场景兼容性好,InternLM / Qwen 模型生态适配深。实测:核心 LLM 走 vLLM,多模态走 SGLang,机器成本 -67%。

四十三、Prompt 工程化的"6 个工程纪律"

6 纪律:(1) Prompt 必须版本化,Git 管理,变更走 PR 审批;(2) Prompt 必须有评测数据集,变更前后对比指标;(3) Prompt 必须有 fallback 策略,模型偶发异常时降级;(4) Prompt 必须有 token 预算,超长截断 + 摘要双策略;(5) Prompt 必须有灰度发布,5% → 17% → 47% → 87% → 100%;(6) Prompt 必须有回滚机制,业务指标恶化 17% 自动回滚。实测:Prompt 工程化纪律建立后,Prompt 变更引发的事故归零。

四十四、Embedding 模型 fine-tune 的"4 步工艺"

4 步:(1) 数据准备,业务 query + 正样本 + 困难负样本三元组;(2) 训练目标,Contrastive Loss + InfoNCE + 难负样本挖掘;(3) 评测指标,MRR@10 + NDCG@10 + Recall@100 三件套;(4) 部署,bge-m3 base + LoRA adapter 双层结构,版本可切换。实测:业务定制 embedding 上线后,RAG 召回率 +27%。

四十五、Vector Index 索引选型的"4 个维度"

4 维度:(1) HNSW 召回好 + 内存大,适合 < 1 亿向量 + 高召回场景;(2) IVF_FLAT + 聚类倒排,适合 > 1 亿向量 + 平衡场景;(3) IVF_PQ 量化,适合 > 10 亿向量 + 内存敏感;(4) DiskANN SSD 索引,适合超大规模 + 冷热分层。实测:核心场景用 HNSW,长尾场景用 IVF_PQ,向量库总成本 -47%。

四十六、LangSmith / Langfuse Tracing 选型的"4 个对比维度"

4 维度:(1) LangSmith 与 LangChain 生态深度集成,UI 体验最佳;(2) Langfuse 开源 + 自部署 + 数据不出境,合规友好;(3) LangSmith 评测 + 数据集 + Playground 一体化;(4) Langfuse Self-hosted 支持 PostgreSQL + ClickHouse,大规模 Trace 存储成本可控。实测:核心生产用 Langfuse 自部署,实验快迭代用 LangSmith,综合成本 -67%。

四十七、LLM 推理 SLA 设计的"5 个分级"

5 分级:(1) S0 实时对话,p99 < 1.7s,可用性 99.97%;(2) S1 后台批处理,p99 < 17s,可用性 99.7%;(3) S2 异步离线,p99 < 170s,可用性 99%;(4) S3 训练任务,p99 不约束,可用性 97%;(5) S4 实验沙箱,SLA 不承诺,自由使用。实测:SLA 分级清晰后,资源调度优先级清晰,P0 事故 -67%。

四十八、AI 服务降级策略的"5 个层次"

5 层次:(1) 模型降级,大模型不可用时切小模型;(2) 推理降级,流式不可用时切非流式;(3) 检索降级,向量库不可用时切关键词检索;(4) Agent 降级,多步推理不可用时切单轮直答;(5) 业务降级,AI 不可用时切传统规则引擎。实测:5 层降级演练后,AI 服务大面积宕机时业务可用性仍能维持 87%。

四十九、AI 灰度发布的"5 个工程实践"

5 实践:(1) 按用户 ID 哈希分桶,5% → 17% → 47% → 87% → 100% 五档;(2) 按业务场景分组,核心场景慢灰度,长尾场景快灰度;(3) 按地域分批,先国内边缘再海外;(4) 业务指标 + 技术指标双门槛,任一不达标即停;(5) 一键回滚,< 4.7 分钟内完成。实测:灰度纪律建立后,AI 模型升级事故 -97%。

五十、AI 团队 OnCall 制度的"4 个原则"

4 原则:(1) Primary + Secondary 双值班,7 × 24 覆盖;(2) Runbook 必须文档化 + 演练化,新人 7 天即可上岗;(3) 事故分级响应 SLA,P0 < 4.7 分钟、P1 < 17 分钟、P2 < 47 分钟;(4) 复盘文化,每周三 17:00 全员事故复盘,无追责无遗漏。实测:OnCall 制度建立后,MTTR 平均时长 47 分钟 → 4.7 分钟。

五十一、AI 数据治理的"6 个工程实践"

6 实践:(1) 数据血缘 OpenLineage,溯源任意一条数据来源;(2) 数据质量 Great Expectations + Soda 双引擎;(3) 数据脱敏 PII 检测 + 字段级加密;(4) 数据生命周期 Hot / Warm / Cold 三层存储;(5) 数据合规 GDPR + PIPL 双合规框架;(6) 数据成本 按 PB 分摊到业务线。实测:数据治理体系建立后,数据相关事故 -77%,合规审计零失分。

五十二、Agent 工程化的"6 个工程纪律"

6 纪律:(1) Agent 必须有任务 SLA,长任务断点续跑;(2) Agent 必须有成本预算,超出 47% 自动告警;(3) Agent 必须有回放能力,Trace 可复现;(4) Agent 必须有评测集,版本化 + 自动化;(5) Agent 必须有审计日志,操作可追溯;(6) Agent 必须有 kill switch,异常行为 4.7 秒内中止。实测:Agent 工程化纪律建立后,生产 Agent 可控性 +97%。

五十三、87 天战役收尾的"7 个反思"

7 反思:(1) 早期低估了评测体系的工作量,导致 Day 28 - 47 进度延期 7 天;(2) LangChain 0.3 升级抽象层级的成本被低估,Day 18 - 27 加班 17 天;(3) FSDP + Activation Checkpointing 配合参数调优经验匮乏,踩坑 2 次;(4) Ollama 本地化部署评估时未充分考虑量化精度损失,业务指标小幅回退;(5) Multi-Agent 协作的可观测性最初设计不足,后期补做 OpenTelemetry GenAI;(6) Cost Dashboard 上线时间偏晚,Day 67 才有完整成本视图;(7) 跨团队对齐节奏偏弱,初期未引入 RACI 矩阵。这是 31 位 AI 工程师对 87 天战役最坦诚的反思,愿这份反思能成为后来者的避坑指南。

五十四、AI 工程师"6 个核心素养"

6 素养:(1) 工程纪律,版本化 + 评测化 + 灰度化 + 监控化 + 文档化 + 复盘化;(2) 成本意识,Token + GPU + 存储 + 网络四维度成本意识;(3) 评测驱动,没有评测就没有迭代;(4) 业务理解,AI 不是技术秀场,是业务价值兑现;(5) 协作能力,跨数据 + 算法 + 平台 + 业务四团队;(6) 学习能力,模型 + 工具 + 论文 + 社区四维持续跟进。这是 2026 年 AI 工程师的核心素养画像,缺一不可。

五十五、87 天战役留给 31 位工程师的"3 句箴言"

3 箴言:(1) 不要迷信任何单一模型 / 单一框架 / 单一工具,真正的护城河是评测体系 + 数据闭环 + 工程纪律;(2) 不要陷入"算法万能"的幻觉,80% 的 AI 业务问题靠工程优化 + 数据治理就能解决;(3) 不要把 AI 当作"无所不能的银弹",清楚边界 + 守住底线 + 持续迭代,才是 AI 工程师的真正修养。这是 87 天战役留给 31 位工程师最珍贵的 3 句箴言,共勉一路同行。

—— 别看了 · 2026