New World

你是一个具有充分实战经验，有着丰富教学与面试别人&准备面试的企业计算机老师，请你向我教学FastAPI，要求：
1. 掌握 FastAPI 基础，能够创建路由、处理请求 
2. 表述简洁准确，必要时给予简洁的代码片段 
3. 能够应对 google，meta，bytedance 等企业的高标准面试 
4. 针对上一点给出 5个左右可能的面试题并给出标准答案 
5. 表述兼具逻辑化，结构化，但不可分点过碎

**大模型应用开发基础 **+ 手撕 Naive RAG

学习内容:

• 后端基础: FastAPI 路由、异步 I/O、Pydantic 数据校验
• LangChain 核心: LLM, Prompt Templates, Output Parsers, LCEL
• Naive RAG 流程: Document Loaders, Text Splitters, Embeddings, Vector Stores
• 向量数据库: FAISS/ChromaDB 本地化使用

手撕系列:

• FastAPI 搭建 “Hello, World” API 服务
• LangChain LCEL 编写第一个 LLM Chain
• 30分钟手撕一个完整的 Naive RAG 应用

解锁技能:

• 熟练使用 FastAPI 搭建 API
• 掌握 LangChain 核心组件与 LCEL 表达式语言
• 能够从零开始，快速构建一个基于文档问答的 RAG Demo

FastAPI

1. FastAPI 介绍

FastAPI 是基于 Python 3.7+ 的现代异步 Web 框架，核心优势三点：

• 性能：底层基于 Starlette（ASGI），性能接近 NodeJS 和 Go，远超 Flask/Django REST
• 开发效率：利用 Python类型检测自动完成参数校验、序列化、文档生成
• 自动文档：内置 Swagger UI (/docs) 和 ReDoc (/redoc)，零额外配置

1.2 核心概念与路由创建

1.2.1 最小可运行应用

from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"message": "Hello, FastAPI"}

启动：uvicorn main:app --reload ; uvicorn 是 ASGI server，--reload 用于开发热重载

1.2.2 路径参数 vs 查询参数

# 路径参数：URL 的一部分，必填
@app.get("/items/{item_id}")
def get_item(item_id: int):  # 自动类型转换+校验
    return {"item_id": item_id}

# 查询参数：URL ?后面，可选
@app.get("/items/")
def list_items(skip: int = 0, limit: int = 10):
    return {"skip": skip, "limit": limit}

FastAPI 根据函数参数是否出现在路径模板中，自动判断它是路径参数还是查询参数，无需任何额外注解

1.2.3 请求体

from pydantic import BaseModel

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False  # 有默认值 = 可选字段

@app.post("/items/")
def create_item(item: Item):
    return item

Pydantic 作用：数据验证、类型转换、JSON Schema 生成。这是 FastAPI 的核心设计哲学。

1.2.4 五种 HTTP 方法

@app.get("/items/{id}")      # 读取
@app.post("/items/")         # 创建
@app.put("/items/{id}")      # 全量更新
@app.patch("/items/{id}")    # 部分更新
@app.delete("/items/{id}")   # 删除

1.2.5 响应模型

class ItemOut(BaseModel):
    name: str
    price: float
    # 故意不包含 secret_field

@app.post("/items/", response_model=ItemOut)
def create_item(item: Item):
    return item  # secret_field 会被自动过滤

response_model 的作用不只是文档，它强制过滤输出字段，是生产环境中防止数据泄露的重要手段

1.3 依赖注入

from fastapi import Depends

def get_db():
    db = SessionLocal()
    try:
        yield db          # yield 版本 = 有 teardown 逻辑
    finally:
        db.close()

def get_current_user(token: str, db = Depends(get_db)):
    user = db.query(User).filter(...).first()
    if not user:
        raise HTTPException(status_code=401)
    return user

@app.get("/profile/")
def profile(user = Depends(get_current_user)):
    return user

依赖可以嵌套、共享、自动处理生命周期。相当于把横切关注点（auth、db session、rate limit）从业务逻辑中彻底解耦

1.4 异步支持

import asyncio

@app.get("/async-data/")
async def get_async_data():
    await asyncio.sleep(1)  # 模拟 I/O（数据库查询、外部 API）
    return {"data": "fetched"}

什么时候用 async def，什么时候用普通 def

场景	建议
数据库查询（async driver）、HTTP 请求	async def
调用同步阻塞库（如 requests）	普通 def（FastAPI 会放到线程池）
CPU 密集型（图像处理、ML 推理）	普通 def + BackgroundTasks 或独立进程

1.5 错误处理

from fastapi import HTTPException

@app.get("/items/{item_id}")
def get_item(item_id: int):
    if item_id not in db:
        raise HTTPException(status_code=404, detail="Item not found")
    return db[item_id]

自定义全局异常处理器：

from fastapi.requests import Request
from fastapi.responses import JSONResponse

@app.exception_handler(ValueError)
async def value_error_handler(request: Request, exc: ValueError):
    return JSONResponse(status_code=400, content={"detail": str(exc)})

1.6 中间件

from fastapi.middleware.cors import CORSMiddleware
import time

# 内置 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_methods=["*"],
)

# 自定义中间件：请求计时
@app.middleware("http")
async def add_process_time(request: Request, call_next):
    start = time.time()
    response = await call_next(request)
    response.headers["X-Process-Time"] = str(time.time() - start)
    return response

1.7 Router 模块化

# routers/items.py
from fastapi import APIRouter
router = APIRouter(prefix="/items", tags=["items"])

@router.get("/")
def list_items(): ...

# main.py
from routers import items
app.include_router(items.router)

生产项目推荐结构：

project/
├── main.py
├── routers/
│   ├── users.py
│   └── items.py
├── models/        # Pydantic / ORM models
├── deps.py        # 公共依赖（db, auth）
└── config.py      # 配置（pydantic-settings）

2. FastAPI 和 Flask 的核心差异是什么？选型依据？

WSGI / ASGI 两个都是协议/接口规范，定义的是 Web 服务器（如 Nginx）和 Python 应用之间怎么传递请求

用户浏览器 → Nginx → [WSGI/ASGI 接口] → Python 应用

你的 Flask/FastAPI 代码不能直接接收网络请求，需要一个中间人（服务器）翻译 HTTP 请求再传给你的代码。WSGI/ASGI 就是这个”翻译协议”

	WSGI	ASGI
全称	Web Server Gateway Interface	Asynchronous Server Gateway Interface
年代	2003年	2019年
支持 async	❌	✅
代表框架	Flask, Django	FastAPI, Starlette
代表服务器	gunicorn, uWSGI	uvicorn, daphne

WSGI 是同步的，意味着一个请求没处理完，这个线程就干不了别的。所以 Flask 要支持并发，只能靠多开线程——来100个请求，就需要100个线程同时跑，线程多了内存消耗大，切换开销也大。

ASGI 一个线程，靠”遇到等待就切换”，处理大量并发，不需要开很多线程，也不需要任何补丁。

高并发请求进来
      ↓
Flask(WSGI)：开多线程扛 → 线程多了撑不住 → gevent 打补丁"假装异步"
      ↓
FastAPI(ASGI)：单线程事件循环，原生 async，遇到 I/O 等待自动切换 → 真正高效

差异集中在三个维度：并发模型、类型系统、开发效率。

Flask 是同步 WSGI 框架，并发依赖多线程或 gevent patch；FastAPI 基于 ASGI，原生 async/await，在高 I/O 并发场景（如数据库、外部 API 调用）下吞吐量显著更高。

FastAPI 强制使用 Type Hints + Pydantic，数据校验、序列化、文档在编写函数签名时同步完成，减少大量样板代码；Flask 需要手动引入 Marshmallow 等库。

选型：若项目是 I/O 密集型、需要自动文档、团队接受 type hint 规范 → FastAPI；若项目遗留代码多、依赖大量同步 Flask 插件生态 → Flask。

3. FastAPI 的依赖注入和普通函数调用有什么本质区别？

本质区别在于生命周期管理和自动解析。

普通函数调用是命令式的，调用者负责创建和销毁资源。Depends 是声明式的——FastAPI 的 DI 系统在请求处理前自动解析依赖图、按顺序执行、在请求结束后自动调用 yield 依赖的清理逻辑（finally 块）。

此外，同一个依赖在同一请求中默认只执行一次（缓存），可通过 use_cache=False 关闭。这对数据库连接复用至关重要。

依赖注入还使单元测试变得简单：可以通过 app.dependency_overrides 在测试中替换真实依赖（如数据库）为 mock。

4. 在 async 中调用了一个同步阻塞的库会发生什么, 如何解决？

async def 路由运行在事件循环的主线程中。若在其中调用同步阻塞操作（如 requests.get()、time.sleep()），会阻塞整个事件循环，导致所有并发请求被挂起，彻底丧失异步优势。

解决方案有两种：

1. 换用异步库：将 requests 替换为 httpx（async 模式）或 aiohttp
2. 用线程池卸载：使用 asyncio.get_event_loop().run_in_executor() ，FastAPI 会自动将其放到线程池执行，不阻塞事件循环。

# 推荐：直接用普通 def，FastAPI 自动处理
@app.get("/sync-task/")
def sync_heavy_task():
    result = some_blocking_library()  # 安全，运行在线程池
    return result

5. 如何在 FastAPI 中实现 JWT 认证？依赖注入如何配合使用?

HTTP 是无状态协议，每次请求服务器都不认识你。所以需要一种机制，让你每次请求都”带着身份证”过来。历史上有两种方案：

方案	原理	问题
Session	服务器存你的登录状态，给你一个 ID	服务器要存储，难以横向扩展
JWT	身份信息直接存在 Token 里给你	服务器不需要存任何东西

JWT本质是个字符串, 分三段，用 . 隔开：

eyJhbGciOiJIUzI1NiJ9.eyJ1c2VyX2lkIjoxMjN9.abc123signature
    ↑ Header（声明用何加密算法）  ↑ Payload（数据）   ↑ Signature（签名）

Signature：用服务器私钥对前两段签名，防止篡改。Payload 是公开的，但没有私钥就无法伪造签名，所以服务器只要验证签名合法，就能信任里面的数据。

整体分三层：Token 签发 → Token 解析 → 路由保护。

① 登录 用户 → POST /login {username, password} → 服务器验证 → 返回 JWT Token

② 携带 Token 请求 用户 → GET /profile Header: Authorization: Bearer → 服务器

③ 服务器验签 服务器拿私钥验证签名 → 合法 → 从 Payload 取出 user_id → 知道你是谁

from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
        user_id = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=401, detail="Invalid token")
    return get_user(user_id)

@app.get("/me/")
async def read_me(user = Depends(get_current_user)):
    return user

OAuth2PasswordBearer 会自动从 Authorization: Bearer <token> 头提取 token，并在 Swagger UI 中生成认证按钮。认证逻辑通过 Depends 注入，所有需要保护的路由只需声明依赖，无需重复代码

6. 如何对 FastAPI 应用做性能优化？从架构到代码层面回答

代码层：确保所有 I/O 操作使用异步驱动（如 asyncpg 代替 psycopg2、motor 代替 pymongo）；避免在 async def 中混用同步阻塞调用；使用 response_model_exclude_unset=True 减少不必要的序列化。

应用层：CPU 密集型任务（ML 推理、图像处理）使用 BackgroundTasks 或推入消息队列（Celery/Redis）异步处理，不阻塞请求响应；合理使用连接池（数据库连接池大小 = worker 数 × 并发量估算）。

部署层：用 gunicorn -k uvicorn.workers.UvicornWorker -w 4 多进程启动，充分利用多核；前置 Nginx 做静态资源缓存和限流；高频幂等接口加 Redis 缓存层。

监控：接入 OpenTelemetry + Prometheus，通过 /metrics 端点暴露请求延迟、错误率，实现可观测性驱动优化，而不是凭感觉调参。

LangChain

7. LangChain介绍

在企业级应用中，我们不再把 LangChain 看作简单的“大模型 API 封装库”，而是把它视作一个面向大模型工作流的声明式编排框架。它的核心基础被严谨地划分为四个基石：标准协议(Runnable)、数据流转(Model I/O)、状态管理(Memory/History) 以及 检索架构(RAG Pipeline)

7.1 大一统的 Runnable 协议与 LCEL

这是现代 LangChain 的灵魂。为了解决旧版本中组件之间互相调用极度混乱、流式输出（Streaming）无法穿透的问题，LangChain 引入了 Runnable 接口。
你可以把Runnable理解为一个标准化的“黑盒”。无论这个黑盒是一个 Prompt、一个大模型、一个输出解析器，还是一个检索器，它们在底层都继承自 Runnable，并强制实现了同一套方法：

• .invoke(input): 同步单次调用
• .stream(input): 流式输出（返回生成器）
• .batch([input1, input2]): 批量处理（底层自动并发）
• 以及它们的异步版本 .ainvoke, .astream, .abatch

LCEL (LangChain Expression Language) 的本质：
当你使用管道符 | 将两个 Runnable 连在一起时（例如 A | B），LCEL 在底层实际上创建了一个 RunnableSequence。它会自动处理数据的输入输出校验，并且智能地透传流式生成——即 A 每吐出一个字，就立刻喂给 B，而不是等 A 全部生成完再传给 B。

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

# 1. 这是一个 Runnable (接收 dict，输出 PromptValue 对象)
prompt = ChatPromptTemplate.from_template("解释这个概念: {concept}")
# 2. 这是一个 Runnable (接收 PromptValue，输出 AIMessage 对象)
model = ChatOpenAI(model="gpt-4o")
# 3. 这是一个 Runnable (接收 AIMessage，提取其中的 content 字符串)
parser = StrOutputParser()
# 拼接形成 RunnableSequence。输入 dict，输出纯字符串。
chain = prompt | model | parser
# 面试关键点：这行代码天然支持高并发批量处理，无需你自己写多线程
results = chain.batch([{"concept": "并发"}, {"concept": "闭包"}])

7.2 Model I/O 的现代化设计

Model I/O 处理的是大模型应用的“咽喉”地带：怎么把人的话结构化地喂给模型，又怎么把模型的话结构化地提取出来。

1. 从 String Prompt 到 Chat Prompt

现代大模型（GPT-4, Claude-3）都是基于角色（Role）的对话模型。我们不再向模型塞一整段长字符串，而是传递一个包含 System, Human, AI 区分的消息列表（Message List）。

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个资深架构师，用{language}回答问题。"),
    ("human", "请解释微服务架构的优缺点。")
])

2. 结构化输出的终极解法：.with_structured_output()

过去，开发者为了让模型输出 JSON，要在 Prompt 里苦口婆心地写“请严格输出 JSON，不要带额外的文字…”。
现在，LangChain 深度整合了各大厂商的 Function Calling / Tool Calling 底层能力。你只需要定义一个 Pydantic 类，LangChain 会自动将其编译为底层的 JSON Schema 传给模型，并自动完成反序列化。

from pydantic import BaseModel, Field

# 定义期望的数据结构
class Person(BaseModel):
    name: str = Field(description="人名")
    age: int = Field(description="年龄")

# 直接在模型上绑定结构，返回的是一个新的 Runnable
structured_model = model.with_structured_output(Person)

# 直接得到严格强类型的 Python 对象，无需自己 parse JSON
person_obj = structured_model.invoke("昨天我见到了28岁的张三。")
print(person_obj.name)  # "张三"
print(person_obj.age)   # 28

7.3 状态管理（记忆）的彻底解耦

面试官常问：“大模型本身是没有记忆的，LangChain 是怎么实现多轮对话的？” 本质上，记忆 = 历史记录提取 + Prompt 动态注入。
在旧版中，ConversationBufferMemory 被死死地绑定在 LLMChain 内部，这导致在构建支持百万用户的 Web 后端时，极难实现基于数据库的分布式存储。

现代架构中，引入了 RunnableWithMessageHistory。它是一个装饰器/中间件，把业务逻辑和存储逻辑彻底切分开来。

from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 模拟一个数据库存储：session_id -> History 对象
store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# 核心业务链（它只管处理对话，不需要知道历史记录存哪）
# 注意：Prompt 中必须留有一个占位符给历史消息（MessagesPlaceholder）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有用的助手。"),
    ("placeholder", "{chat_history}"),  # 历史消息会注入到这里
    ("human", "{input}"),
])
chain = prompt | model | parser

# 注入历史记录的“包装器”
chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# 调用时，必须传入配置告知 session_id
chain_with_history.invoke(
    {"input": "我叫Alice"},
    config={"configurable": {"session_id": "user_123"}}
)

7.4 RAG 组装函数的变迁

在 RAG（检索增强生成）场景下，我们通常需要先查向量库，再组装 Prompt，再调模型。
旧版的 RetrievalQA 已经被官方强烈建议弃用。现在官方提供了更底层的“工厂函数”来拼装 LCEL 链：

1. **create_stuff_documents_chain**：负责把检索到的多篇文档（Documents）“塞”（stuff）进 Prompt 中。
2. **create_retrieval_chain**：负责协调外围流程：拿用户问题 -> 调 Retriever -> 把文档交给上面的文档链 -> 输出最终结果。

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

# 1. 创建生成答案的内部链
system_prompt = (
    "你是一个问答助手。请使用以下检索到的上下文来回答问题。\n"
    "上下文: {context}"
)
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", system_prompt),
    ("human", "{input}"),
])
question_answer_chain = create_stuff_documents_chain(model, qa_prompt)

# 2. 将检索器与上面的内部链组合成最终链
# 假设 retriever 已经初始化好 (如 retriever = vectorstore.as_retriever())
rag_chain = create_retrieval_chain(retriever, question_answer_chain)

# 调用时，它会自动在内部执行检索，并将 Document 列表注入到 {context} 变量中
response = rag_chain.invoke({"input": "公司的请假流程是什么？"})
print(response["answer"])

这种分离带来的巨大好处是：如果你想在中途加入一个“重排查（Reranker）”节点，或者想在拿到答案后进行“涉黄涉暴校验”，你可以直接通过 LCEL 轻松地在中间插入对应的 Runnable 节点，这在以前的黑盒 API 中是根本做不到的。

Advanced RAG 与生产级向量数据库

学习内容:

• Advanced RAG 技术: Query Transformation, Re-ranking, Hybrid Search
• RAG 评估: 使用 RAGAs, TruLens 进行自动化评估
• 生产级向量数据库: Milvus/Zilliz Cloud 部署与使用
• 数据处理: Unstructured.io 解析复杂文档

手撕系列:

• 实现 BM25 + 向量的混合检索
• 引入 Cohere Rerank 模型提升检索精度
• 使用 RAGAs 评估 RAG 系统的 Faithfulness 和 Answer Relevancy
• Docker 部署 Milvus 并进行增删改查操作

解锁技能:

• 掌握 10+ 种 RAG 优化策略
• 能够建立 RAG 系统的自动化评估流水线
• 熟练使用生产级的分布式向量数据库 Milvus
• 具备处理复杂、非结构化文档的能力

Agent 开发与 Tool Calling

学习内容:

• Agent 核心: ReAct 框架, Planning, Tool Use, Memory
• Tool Calling: OpenAI Function Calling, Tool Schema 定义
• 工具开发: 如何将 API, 数据库查询等封装为 Agent 可用的工具
• 错误处理: 工具调用失败的重试、降级策略

手撕系列:

• 实现 3个 自定义工具 (天气查询, SQL数据库查询, API调用)
• 基于 LangChain 构建一个可以链式调用工具的 Agent
• 使用 OpenAI Function Calling 实现结构化数据提取

解锁技能:

• 深刻理解 Agent 的”思考-行动”工作流
• 能够开发、测试、维护自定义工具集
• 掌握 Function Calling 的原理与应用
• 具备构建能处理真实世界任务的 Agent 的能力

系统性能优化

学习内容:

• 缓存策略: Redis 缓存 LLM 响应和 Embedding 结果
• 异步处理: asyncio, aiohttp 实现高并发
• 批处理优化: Embedding 和 LLM 调用的批处理
• 推理加速: vLLM, TensorRT-LLM 部署与使用

手撕系列:

• 为 RAG 系统引入 Redis 缓存，对比优化前后性能
• 将 FastAPI 的同步接口改造为异步接口
• 部署 vLLM 并通过 API 进行推理

解锁技能:

• 掌握 LLM 应用的核心性能优化手段
• 能够将系统的 QPS 提升 10 倍以上
• 熟练使用 Redis 进行缓存设计
• 具备部署和使用高性能推理引擎的能力

监控、可观测性与部署

学习内容:

• Agent 链路追踪: LangSmith, OpenTelemetry
• 指标监控: Prometheus 监控业务和系统指标
• 可视化: Grafana 创建监控大盘
• 日志系统: ELK Stack (Elasticsearch, Logstash, Kibana)
• 容器化部署: Docker, Docker Compose

手撕系列:

• 为 Agent 应用集成 LangSmith，追踪每一步的调用和延迟
• 使用 Prometheus 暴露自定义指标 (如 Token 消耗, 缓存命中率)
• 使用 Docker Compose 将 FastAPI + Milvus + Redis 整套系统一键部署

解锁技能:

• 具备构建完整 LLM 应用可观测性体系的能力
• 能够快速定位和诊断线上问题
• 掌握基于 Docker 的容器化部署和编排
• 拥有完整的 DevOps for LLM Apps 经验

Multi-Agent 系统开发

学习内容:

• Multi-Agent 框架: AutoGen vs. CrewAI
• Agent 角色定义: 如何设计具有不同职责和能力的 Agent
• 通信机制与工作流: GroupChat, Sequential/Hierarchical flow
• 状态管理: 如何在多个 Agent 之间共享和传递状态

手撕系列:

• 使用 AutoGen 构建一个“研究员-程序员-测试员”协作的软件开发团队
• 使用 CrewAI 构建一个“旅行规划师-本地向导-预订专员”的旅行 Agent

解锁技能:

• 掌握至少两种主流的 Multi-Agent 开发框架
• 能够根据复杂业务需求，设计和实现多智能体协作系统
• 理解不同协作模式 (如层级式 vs. 对话式) 的优缺点

工业级项目实战与面试准备