Back to Details

server-skills

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Server Skills for LlamaFarm

LlamaFarm 服务器端技能指南

Framework-specific patterns and code review checklists for the LlamaFarm Server component.
针对LlamaFarm服务器组件的框架特定设计模式与代码评审检查清单。

Overview

概览

PropertyValue
Path
server/
Python3.12+
FrameworkFastAPI 0.116+
Task QueueCelery 5.5+
ValidationPydantic 2.x, pydantic-settings
Loggingstructlog with FastAPIStructLogger
属性取值
路径
server/
Python版本3.12+
框架FastAPI 0.116+
任务队列Celery 5.5+
校验工具Pydantic 2.x, pydantic-settings
日志工具基于FastAPIStructLogger的structlog

Links to Shared Skills

共享技能链接

This skill extends the shared Python skills. See:
  • Python Patterns - Dataclasses, comprehensions, imports
  • Async Patterns - async/await, asyncio, concurrency
  • Typing Patterns - Type hints, generics, Pydantic
  • Testing Patterns - Pytest, fixtures, mocking
  • Error Handling - Exceptions, logging, context managers
  • Security Patterns - Path traversal, injection, secrets
此技能扩展了共享的Python技能。请查看：
  • Python设计模式 - 数据类、推导式、导入规范
  • 异步设计模式 - async/await、asyncio、并发处理
  • 类型标注设计模式 - 类型提示、泛型、Pydantic用法
  • 测试设计模式 - Pytest、fixtures、模拟测试
  • 错误处理 - 异常、日志、上下文管理器
  • 安全设计模式 - 路径遍历、注入攻击、密钥管理

Server-Specific Checklists

服务器端专属检查清单

TopicFileKey Points
FastAPIfastapi.mdRoutes, dependencies, middleware, exception handlers
Celerycelery.mdTask patterns, error handling, retries, signatures
Pydanticpydantic.mdPydantic v2 models, validation, serialization
Performanceperformance.mdAsync patterns, caching, connection pooling
主题文件核心要点
FastAPIfastapi.md路由、依赖项、中间件、异常处理器
Celerycelery.md任务模式、错误处理、重试机制、签名用法
Pydanticpydantic.mdPydantic v2模型、校验规则、序列化
性能优化performance.md异步模式、缓存、连接池

Architecture Overview

架构概览

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
server/
├── main.py                 # Uvicorn入口文件，MCP挂载点
├── api/
│   ├── main.py             # FastAPI应用工厂、中间件配置
│   ├── errors.py           # 自定义异常 + 异常处理器
│   ├── middleware/         # ASGI中间件（structlog、错误处理）
│   └── routers/            # API路由模块
│       ├── projects/       # 项目CRUD端点
│       ├── datasets/       # 数据集管理
│       ├── rag/            # RAG查询端点
│       └── ...
├── core/
│   ├── settings.py         # pydantic-settings配置
│   ├── logging.py          # structlog配置、FastAPIStructLogger
│   └── celery/             # Celery应用配置
│       ├── celery.py       # Celery应用实例
│       └── rag_client.py   # RAG任务签名与辅助工具
├── services/               # 业务逻辑层
│   ├── project_service.py  # 项目CRUD操作
│   ├── dataset_service.py  # 数据集管理
│   └── ...
├── agents/                 # AI Agent实现
└── tests/                  # Pytest测试套件

Quick Reference

快速参考

Settings Pattern (pydantic-settings)

配置模式（pydantic-settings）

python
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
python
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()  # 模块级单例

Structured Logging

结构化日志

python
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
python
from core.logging import FastAPIStructLogger

logger = FastAPIStructLogger(__name__)
logger.info("操作完成", extra={"count": 10, "duration_ms": 150})
logger.bind(namespace=namespace, project=project_id)  # 添加上下文

Custom Exceptions

自定义异常

python
undefined
python
undefined

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")
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"项目 {namespace}/{project_id} 不存在")

Register handler in api/errors.py

在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)
undefined
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)
undefined

Service Layer Pattern

服务层模式

python
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
python
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)
        # ... 加载与校验

Review Checklist Summary

检查清单摘要

  1. FastAPI Routes (High priority)
    • Proper async/sync function choice
    • Response model defined with 
      response_model=
    • OpenAPI metadata (operation_id, tags, summary)
    • HTTPException with proper status codes
  2. Celery Tasks (High priority)
    • Use signatures for cross-service calls
    • Implement proper timeout and polling
    • Handle task failures gracefully
    • Store group metadata for parallel tasks
  3. Pydantic Models (Medium priority)
    • Use Pydantic v2 patterns (model_config, Field)
    • Proper validation with field constraints
    • Serialization with model_dump()
  4. Performance (Medium priority)
    • Avoid blocking calls in async functions
    • Use proper connection pooling for external services
    • Implement caching where appropriate
See individual topic files for detailed checklists with grep patterns.
  1. FastAPI路由（高优先级）
    • 合理选择异步/同步函数
    • 通过
      response_model=
      定义响应模型
    • 配置OpenAPI元数据（operation_id、标签、摘要）
    • 使用带正确状态码的HTTPException
  2. Celery任务（高优先级）
    • 使用签名实现跨服务调用
    • 配置合理的超时与轮询机制
    • 优雅处理任务失败
    • 为并行任务存储组元数据
  3. Pydantic模型（中优先级）
    • 使用Pydantic v2设计模式（model_config、Field）
    • 通过字段约束实现合理校验
    • 使用model_dump()进行序列化
  4. 性能优化（中优先级）
    • 避免在异步函数中使用阻塞调用
    • 为外部服务配置合理的连接池
    • 在合适场景实现缓存
请查看各主题文件获取包含grep模式的详细检查清单。