企业官网建设流程全解析

# AI Engineering 2026路线图：RAG、Agent与MCP实战

## 1. 背景：从“调API”到“造系统”的鸿沟

2026年，LLM API已不再是门槛。调用GPT-4o或Claude 3.5只需几行Python代码，但真正能**部署、监控、迭代**生产级AI系统的开发者仍然稀缺。AI Engineering作为一门独立学科，正从“数据科学”和“软件工程”的交叉地带独立出来——它不要求你训练模型（ML不再是必选项），但要求你掌握**RAG管道、多智能体协作、MCP协议集成、云原生部署**等全栈能力。

根据Technovids发布的《AI Engineering Roadmap 2026》，一个合格的AI工程师需要在3-6个月（结构化培训）或9-18个月（自学）内掌握：Python、REST APIs、LLM APIs、RAG、向量数据库、LangChain、LangGraph、Agent、MCP、FastAPI、Docker。这些不是孤立的工具，而是一条完整的“从数据到生产”的链路。

本文将沿着这条路线图，深入剖析每个环节的技术原理，并给出可复现的代码示例——一个基于LangChain 0.3.x + ChromaDB + FastAPI的RAG知识助手，覆盖从本地文档到REST API部署的全流程。

## 2. 技术原理：AI工程的核心架构

### 2.1 组件分层

AI工程系统可以抽象为四层：

| 层次 | 技术栈 | 关键职责 |

|------|--------|----------|

| 数据层 | 向量数据库（ChromaDB、Pinecone 5.0）、文档解析器 | 存储非结构化数据的嵌入向量，支持语义检索 |

| 编排层 | LangChain 0.3、LangGraph、CrewAI | 管理LLM调用链、状态图、多智能体协作 |

| 协议层 | MCP（Model Context Protocol）、REST API | 标准化模型与外部工具/数据的交互 |

| 部署层 | FastAPI 0.111、Docker 24.0、LangSmith | 提供HTTP接口，容器化，监控与可观测性 |

### 2.2 RAG：从“检索”到“生成”的闭环

RAG（Retrieval-Augmented Generation）是AI工程的基石。其流程为：

1. **文档分块**：将PDF、Markdown等拆分成固定大小的chunks（如500 tokens）。

2. **嵌入与索引**：用text-embedding-3-small（OpenAI）或BGE（本地）将chunks向量化，存入向量数据库。

3. **检索**：用户查询时，同样嵌入查询向量，通过余弦相似度召回top-k chunks。

4. **增强生成**：将召回的chunks作为上下文拼入Prompt，喂给LLM生成最终回答。

### 2.3 Agent：从“单一流水线”到“自主决策”

LangGraph 0.2.x引入了**状态图**思想，让Agent可以维护记忆、循环、条件分支。例如一个多Agent研究系统：一个Agent负责搜索，一个负责摘要，一个负责验证，通过CrewAI协调。

### 2.4 MCP：工具与模型的标准化桥梁

MCP（Model Context Protocol）是2025年Anthropic提出的开放协议，类似OpenAI的Functions but 跨模型。它定义了工具声明、调用、响应的统一格式。AI工程师可以直接用MCP服务器暴露数据库、文件系统、API等，让任何LLM通过同一协议调用。

## 3. 实践：构建一个可部署的RAG知识助手

下面是一个完整的RAG助手实现，涵盖：文档加载、分块、嵌入、向量存储、检索、生成、FastAPI部署。

```python

# 文件名: rag_assistant.py

# 依赖: langchain==0.3.0, chromadb==0.5.0, fastapi==0.111.0, uvicorn==0.29.0

# 安装: pip install langchain chromadb fastapi uvicorn openai

import os

from fastapi import FastAPI, HTTPException

from pydantic import BaseModel

from langchain_community.document_loaders import TextLoader

from langchain.text_splitter import RecursiveCharacterTextSplitter

from langchain_openai import OpenAIEmbeddings, ChatOpenAI

from langchain_chroma import Chroma

from langchain.chains import RetrievalQA

from langchain.prompts import PromptTemplate

# ---------- 初始化 ----------

app = FastAPI(title="RAG Knowledge Assistant", version="1.0.0")

os.environ["OPENAI_API_KEY"] = "sk-xxx" # 请替换为真实Key

# 持久化向量数据库路径

CHROMA_PATH = "./chroma_db"

# 文档目录（可替换为任意文件夹）

DOCS_DIR = "./docs"

# ---------- 文档预处理 ----------

def index_documents():

"""加载文档目录下的所有txt文件并建立向量索引"""

loaders = []

for fname in os.listdir(DOCS_DIR):

if fname.endswith(".txt"):

loaders.append(TextLoader(os.path.join(DOCS_DIR, fname)))

if not loaders:

return

docs = []

for loader in loaders:

docs.extend(loader.load())

# 分块：每块500字符，重叠50

splitter = RecursiveCharacterTextSplitter(

chunk_size=500, chunk_overlap=50, separators=["\n\n", "\n", "。", "！", "？"]

)

chunks = splitter.split_documents(docs)

# 嵌入模型（OpenAI）

embedding = OpenAIEmbeddings(model="text-embedding-3-small")

# 创建或覆盖索引

vectordb = Chroma.from_documents(

documents=chunks,

embedding=embedding,

persist_directory=CHROMA_PATH

)

vectordb.persist()

print(f"索引完成，共 {len(chunks)} 个文档块")

# 如果索引不存在，则初始化

if not os.path.exists(CHROMA_PATH):

index_documents()

# 加载已有索引

embedding = OpenAIEmbeddings(model="text-embedding-3-small")

vectordb = Chroma(

persist_directory=CHROMA_PATH,

embedding_function=embedding

)

# ---------- 检索增强生成 ----------

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)

prompt_template = """你是一个知识助手，请根据以下上下文回答用户问题。

如果上下文不足以回答问题，请说“未找到相关信息”。

上下文：

{context}

问题：{question}

请回答："""

PROMPT = PromptTemplate(template=prompt_template, input_variables=["context", "question"])

qa_chain = RetrievalQA.from_chain_type(

llm=llm,

chain_type="stuff",

retriever=vectordb.as_retriever(search_kwargs={"k": 3}),

return_source_documents=True,

chain_type_kwargs={"prompt": PROMPT}

)

# ---------- FastAPI 接口 ----------

class QueryRequest(BaseModel):

question: str

class QueryResponse(BaseModel):

answer: str

sources: list[str]

@app.post("/query", response_model=QueryResponse)

async def query(request: QueryRequest):

try:

result = qa_chain.invoke({"query": request.question})

answer = result["result"]

sources = [doc.metadata["source"] for doc in result["source_documents"]]

return QueryResponse(answer=answer, sources=sources)

except Exception as e:

raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")

async def health():

return {"status": "ok", "chunks": vectordb._collection.count()}

# 启动：uvicorn rag_assistant:app --reload --port 8000

```

### 代码说明

- **版本号**：LangChain 0.3.0（2025年底发布，支持最新Chroma集成）、FastAPI 0.111.0、Chroma 0.5.0。

- **分块策略**：RecursiveCharacterTextSplitter，按中文标点分隔，避免断句错误。

- **检索**：返回top-3块，通过`stuff`方式一次性注入Prompt（适合短上下文）。

- **部署**：使用FastAPI暴露`/query`和`/health`端点，Docker打包后即可上线。

### 运行与测试

```bash

# 1. 准备文档目录

mkdir -p docs

echo "AI Engineering（人工智能工程）是使用大语言模型构建生产级系统的学科。" > docs/intro.txt

echo "RAG（检索增强生成）通过外部知识库提升LLM回复准确性。" > docs/rag.txt

# 2. 启动服务

uvicorn rag_assistant:app --reload --port 8000

# 3. 测试

curl -X POST "http://localhost:8000/query" \

-H "Content-Type: application/json" \

-d '{"question": "什么是AI Engineering？"}'

# 返回：{"answer":"AI Engineering（人工智能工程）是使用大语言模型构建生产级系统的学科。","sources":["docs/intro.txt"]}

```

## 4. 进阶：从单Agent到多Agent与MCP

在RAG助手基础上，AI工程师还需要掌握：

### 4.1 LangGraph状态机

```python

from langgraph.graph import StateGraph, END

from typing import TypedDict, List

class AgentState(TypedDict):

messages: List[dict]

tool_results: dict

# 节点函数

def search_node(state: AgentState):

# 调用向量检索或网络搜索

return {"tool_results": {"search": "result..."}}

def response_node(state: AgentState):

# 结合检索结果生成最终回答

return {"messages": [{"role": "assistant", "content": "final answer"}]}

# 构建图

graph = StateGraph(AgentState)

graph.add_node("search", search_node)

graph.add_node("respond", response_node)

graph.set_entry_point("search")

graph.add_edge("search", "respond")

graph.add_edge("respond", END)

```

### 4.2 MCP协议集成

MCP服务器示例（基于FastMCP库，v0.3）：

```python

from fastmcp import FastMCP, Tool

mcp = FastMCP("SQL-DB-Tool")

@mcp.tool()

def query_database(sql: str) -> list:

"""执行SQL查询并返回结果"""

# 实际连接数据库代码

return [{"row1": "data"}]

# 启动MCP服务器

mcp.run(host="0.0.0.0", port=9000)

```

任何支持MCP的LLM（如Claude 3.5 via API）可以直接调用这个工具，无需手动写Function Calling。

## 5. 总结：2026年AI工程师的技能树与路线

从素材中我们可以提炼出AI Engineering的七大全栈项目，这些项目覆盖了从入门到高级的完整路径：

| 项目名称 | 核心技术 | 难度 |

|----------|----------|------|

| 1. RAG知识助手 | LangChain + ChromaDB + FastAPI | 入门 |

| 2. AI简历筛选器 | LLM + 结构化输出（Pydantic） | 入门 |

| 3. 多Agent研究者 | CrewAI + LangGraph + 搜索API | 中级 |

| 4. MCP助手 | FastMCP + 工具链 | 中级 |

| 5. 支持聊天机器人 | 记忆 + 对话管理 | 中级 |

| 6. 工作流自动化 | LangGraph + 判断循环 | 高级 |

| 7. 部署API + 监控 | Docker + LangSmith + 可观测性 | 高级 |

**关键结论**：

- **不需要ML训练**：AI工程完全基于预训练模型API，核心在于系统集成与工程优化。

- **学习路径**：Python → REST API → LLM API → RAG → 向量数据库 → LangChain → Agent → MCP → 部署。建议3个月集中突破。

- **工具选择**：LangChain仍是编排首选（0.3.x），CrewAI适合多Agent协作，LlamaIndex在文档索引方面更细致，但RAG场景LangChain+Chroma已足够。

- **值得关注的版本**：FastAPI 0.111（2025年大版本，支持Pydantic v2）、Docker 24.0（2023年稳定版）、LangSmith用于生产监控。

最后，AI Engineering不是框架的简单堆砌，而是**系统思维**的体现。从本地原型到生产系统，差的不只是几个工具，而是对**延迟、令牌成本、检索质量、错误处理、版本管理**的工程化把控。2026年的路线图已经清晰，剩下的就是动手写代码——从上面的RAG助手开始。

---

*参考：Technovids AI Engineering Roadmap 2026, LangChain 0.3.0 Release Notes, FastAPI 0.111.0 Changelog*