server-skills

LlamaFarm 服务端技能

适用于 LlamaFarm 服务端组件的框架特定模式和代码审查清单。

概述

属性	值
路径	server/
Python	3.12+
框架	FastAPI 0.116+
任务队列	Celery 5.5+
验证	Pydantic 2.x，pydantic-settings
日志记录	使用 FastAPIStructLogger 的 structlog

共享技能链接

此技能扩展了共享的 Python 技能。请参阅：

• Python 模式 - Dataclass、推导式、导入
• 异步模式 - async/await、asyncio、并发
• 类型标注模式 - 类型提示、泛型、Pydantic
• 测试模式 - Pytest、夹具、模拟
• 错误处理 - 异常、日志记录、上下文管理器
• 安全模式 - 路径遍历、注入、机密信息

服务端专用清单

主题	文件	要点
FastAPI	fastapi.md	路由、依赖项、中间件、异常处理器
Celery	celery.md	任务模式、错误处理、重试、签名
Pydantic	pydantic.md	Pydantic v2 模型、验证、序列化
性能	performance.md	异步模式、缓存、连接池

架构概述

server/
├── main.py                 # Uvicorn entry point, MCP mount
├── api/
│   ├── main.py             # FastAPI app factory, middleware setup
│   ├── errors.py           # Custom exceptions + exception handlers
│   ├── middleware/         # ASGI middleware (structlog, errors)
│   └── routers/            # API route modules
│       ├── projects/       # Project CRUD endpoints
│       ├── datasets/       # Dataset management
│       ├── rag/            # RAG query endpoints
│       └── ...
├── core/
│   ├── settings.py         # pydantic-settings configuration
│   ├── logging.py          # structlog setup, FastAPIStructLogger
│   └── celery/             # Celery app configuration
│       ├── celery.py       # Celery app instance
│       └── rag_client.py   # RAG task signatures and helpers
├── services/               # Business logic layer
│   ├── project_service.py  # Project CRUD operations
│   ├── dataset_service.py  # Dataset management
│   └── ...
├── agents/                 # AI agent implementations
└── tests/                  # Pytest test suite

快速参考

设置模式（pydantic-settings）

from pydantic_settings import BaseSettings

class Settings(BaseSettings, env_file=".env"):
    HOST: str = "0.0.0.0"
    PORT: int = 14345
    LOG_LEVEL: str = "INFO"

settings = Settings()  # Module-level singleton

结构化日志记录

from core.logging import FastAPIStructLogger

logger = FastAPIStructLogger(__name__)
logger.info("Operation completed", extra={"count": 10, "duration_ms": 150})
logger.bind(namespace=namespace, project=project_id)  # Add context

自定义异常

# Define exception hierarchy
class NotFoundError(Exception): ...
class ProjectNotFoundError(NotFoundError):
    def __init__(self, namespace: str, project_id: str):
        self.namespace = namespace
        self.project_id = project_id
        super().__init__(f"Project {namespace}/{project_id} not found")

# Register handler in api/errors.py
async def _handle_project_not_found(request: Request, exc: Exception) -> Response:
    payload = ErrorResponse(error="ProjectNotFound", message=str(exc))
    return JSONResponse(status_code=404, content=payload.model_dump())

def register_exception_handlers(app: FastAPI) -> None:
    app.add_exception_handler(ProjectNotFoundError, _handle_project_not_found)

服务层模式

class ProjectService:
    @classmethod
    def get_project(cls, namespace: str, project_id: str) -> Project:
        project_dir = cls.get_project_dir(namespace, project_id)
        if not os.path.isdir(project_dir):
            raise ProjectNotFoundError(namespace, project_id)
        # ... load and validate

审查清单摘要

1. FastAPI 路由（高优先级）

   • 选择合适的异步/同步函数
   • 使用 response_model= 定义响应模型
   • OpenAPI 元数据（operation_id、tags、summary）
   • 使用适当状态码的 HTTPException

2. Celery 任务（高优先级）

   • 跨服务调用时使用签名
   • 实现适当的超时和轮询
   • 妥善处理任务失败
   • 为并行任务存储组元数据

3. Pydantic 模型（中优先级）

   • 使用 Pydantic v2 模式（model_config、Field）
   • 使用字段约束进行适当验证
   • 使用 model_dump() 进行序列化

4. 性能（中优先级）

   • 避免在异步函数中执行阻塞调用
   • 为外部服务使用适当的连接池
   • 在适当场景下实现缓存

请参阅各主题文件，查看包含 grep 模式的详细清单。