标签: 进阶 - 鱼子酱

2025-03-24发表2025-03-24更新 Ethan Wilkins AI / RAG4 分钟读完 (大约559个字)

入门RAG（检索增强生成）

入门 RAG（检索增强生成）可以按照以下步骤进行：

1. 理解 RAG 的基本概念

RAG 是结合 检索（Retrieval） 和 生成（Generation） 的方法，提升大语言模型的准确性和知识覆盖度。
其核心思想是：先从外部知识库中检索相关信息，再将检索结果与输入结合，最后用生成式模型回答问题。

2. 学习 RAG 的关键组件

检索器（Retriever）：负责从数据库、向量存储或文档集中找到最相关的信息。（常见工具：FAISS、ChromaDB）
生成模型（Generator）：利用检索到的信息生成回答。（常见模型：LLaMA、GPT、T5 等）
知识存储（Knowledge Base）：存放外部数据，可以是向量数据库（如 Pinecone、Weaviate）或传统数据库（如 PostgreSQL）。

3. 搭建基础 RAG 流程

可以使用 LangChain 或 LlamaIndex 这类框架快速上手：

数据预处理：清理文本数据，并转换成向量格式。
向量存储：使用 FAISS 或 Pinecone 存储数据。
检索实现：基于相似度搜索获取相关内容。
生成回答：结合检索信息，调用 LLM 生成答案。

4. 实践 RAG 项目

简单实验：用 OpenAI + FAISS 实现基础 RAG。
强化检索能力：尝试使用更好的嵌入模型（如 BGE、Cohere Embed）。
应用优化：加入 Rerank、混合搜索（BM25 + 向量搜索）等。
构建自己的知识库：从 PDF、网页、数据库导入数据，实现企业级 RAG。

5. 进阶探索

深入 LangChain / LlamaIndex，学习更高级的 RAG 管道优化。
优化检索策略，如 Dense Retrieval + Sparse Retrieval 结合。
高效推理，用 LoRA、量化技术降低计算成本。
增强生成稳定性，如 CoT（思维链）、RAG-fusion 等方法。

如果你想快速上手，我建议先用 LangChain + FAISS + OpenAI API 跑一个最小可行的 RAG，然后再逐步优化。你对哪个部分更感兴趣？

2025-03-20发表2025-03-20更新 Ethan Wilkins 网页技术 / Fastapi7 分钟读完 (大约1047个字)

FastAPI高阶用法

FastAPI 是一个现代、高性能的 Python Web 框架，基于 Starlette 和 Pydantic，特别适合构建 RESTful API。以下是一些 FastAPI 的高阶用法，帮助你充分发挥其潜力：

1. 依赖注入 (Dependency Injection)

FastAPI 提供了强大的依赖注入系统，可以在路径操作函数中复用代码、管理资源或执行初始化操作。

from fastapi import Depends, FastAPI

app = FastAPI()

# 定义一个依赖
async def common_parameters(q: str | None = None, skip: int = 0, limit: int = 10):
    return {"q": q, "skip": skip, "limit": limit}

# 在路径操作中使用依赖
@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return {"message": "Hello Items!", "params": commons}

优点: 可以将公共逻辑（如认证、数据库连接）提取到依赖中，保持代码 DRY（Don’t Repeat Yourself）。
进阶用法: 支持嵌套依赖、全局依赖（通过 app.dependencies）。

2. 自定义响应模型与动态校验

通过 Pydantic 模型，你可以定义复杂的输入输出校验逻辑，并动态调整返回内容。

from pydantic import BaseModel, validator
from fastapi import FastAPI

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

    @validator("price")
    def price_must_be_positive(cls, value):
        if value <= 0:
            raise ValueError("Price must be positive")
        return value

@app.post("/items/", response_model=Item)
async def create_item(item: Item):
    return item

高阶技巧: 使用 response_model_exclude_unset=True 排除未设置的字段，或者用 Union 类型支持多种返回模型。

3. 异步数据库操作

FastAPI 支持异步编程，结合 ORM（如 SQLAlchemy 的异步支持或 Tortoise ORM）可以显著提升性能。

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

app = FastAPI()

engine = create_async_engine("sqlite+aiosqlite:///example.db")
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)

async def get_db():
    async with AsyncSessionLocal() as session:
        yield session

@app.get("/data/")
async def read_data(db: AsyncSession = Depends(get_db)):
    # 异步查询数据库
    result = await db.execute("SELECT * FROM some_table")
    return result.fetchall()

注意: 确保数据库驱动支持异步（如 aiosqlite 或 asyncpg）。

4. 中间件 (Middleware)

FastAPI 支持自定义中间件，用于处理请求和响应的全局逻辑，例如日志、CORS 或认证。

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time

app = FastAPI()

# 自定义中间件
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# 添加 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

用途: 日志记录、请求修改、性能监控等。

5. WebSocket 支持

FastAPI 支持 WebSocket，适合实时应用场景。

from fastapi import FastAPI, WebSocket

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Message received: {data}")

高阶用法: 结合依赖注入管理 WebSocket 连接状态，或实现广播功能。

6. 背景任务 (Background Tasks)

对于耗时操作（如发送邮件、处理文件），可以使用背景任务异步执行。

from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

def write_log(message: str):
    with open("log.txt", "a") as f:
        f.write(f"{message}\n")

@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks):
    background_tasks.add_task(write_log, f"Notification sent to {email}")
    return {"message": "Notification queued"}

注意: 背景任务不会阻塞响应，但需要确保任务本身是线程安全的。

7. 自定义异常处理

FastAPI 允许你定义全局异常处理器，统一处理特定类型的错误。

from fastapi import FastAPI, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

app = FastAPI()

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request, exc):
    return JSONResponse(
        status_code=400,
        content={"message": "Validation error", "details": exc.errors()},
    )

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    if item_id == 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item_id": item_id}

进阶: 可以为特定异常类（如数据库错误）定义独立的处理器。

8. OpenAPI 扩展

FastAPI 自动生成 OpenAPI 文档，你可以通过 tags、responses 等参数增强文档。

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/", tags=["items"], summary="Get all items", response_description="List of items")
async def read_items():
    return [{"id": 1, "name": "Foo"}]

高阶技巧: 使用 openapi_extra 参数自定义 OpenAPI Schema，或集成外部工具（如 ReDoc）优化文档展示。