Roog (顾新培)
8ca2f64f7e
变更内容： - 删除 `src` 子目录，将模块引用路径从 `src.functional_scaffold` 更新为 `functional_scaffold`。 - 修改相关代码、文档、测试用例及配置文件中的路径引用，包括 `README.md`、`Dockerfile`、`uvicorn` 启动命令等。 - 优化项目目录结构，提升代码维护性和可读性。
5.3 KiB
5.3 KiB
快速入门指南
本指南帮助算法同学快速上手 FunctionalScaffold 脚手架，在 10 分钟内完成第一个算法服务的开发和部署。
核心理念
算法同学只需关注核心算法逻辑，框架自动处理：
- HTTP 接口封装
- 参数验证
- 错误处理
- 日志记录
- 性能指标
- 健康检查
- 容器化部署
环境准备
1. 安装依赖
# 克隆项目
git clone <repository-url>
cd FunctionalScaffold
# 创建虚拟环境
python -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 安装依赖
pip install -e ".[dev]"
2. 启动服务
# 开发模式（自动重载）
uvicorn functional_scaffold.main:app --reload --port 8000
# docker 开发者模式
cd deployment && docker compose up -d
3. 验证服务
# 健康检查
curl http://localhost:8000/healthz
# 调用示例算法
curl -X POST http://localhost:8000/invoke \
-H "Content-Type: application/json" \
-d '{"number": 17}'
添加你的第一个算法
步骤 1：创建算法类
在 src/functional_scaffold/algorithms/ 目录下创建新文件：
# src/functional_scaffold/algorithms/my_algorithm.py
from typing import Dict, Any
from .base import BaseAlgorithm
class MyAlgorithm(BaseAlgorithm):
"""我的算法类"""
def process(self, input_data: Any) -> Dict[str, Any]:
"""
算法处理逻辑 - 只需实现这个方法！
Args:
input_data: 输入数据
Returns:
Dict[str, Any]: 处理结果
"""
# 在这里实现你的算法逻辑
result = self._do_calculation(input_data)
return {
"input": input_data,
"output": result,
"message": "处理成功"
}
def _do_calculation(self, data):
"""内部计算方法"""
# 你的算法实现
return data * 2
步骤 2：注册算法
在 src/functional_scaffold/algorithms/__init__.py 中添加导出：
from .base import BaseAlgorithm
from .prime_checker import PrimeChecker
from .my_algorithm import MyAlgorithm # 添加这行
__all__ = ["BaseAlgorithm", "PrimeChecker", "MyAlgorithm"] # 添加到列表
步骤 3：添加 API 端点
在 src/functional_scaffold/api/routes.py 中添加路由：
from ..algorithms.my_algorithm import MyAlgorithm
@router.post("/my-endpoint")
async def my_endpoint(
request: MyRequest, # 需要定义请求模型
request_id: str = Depends(get_request_id)
):
"""我的算法端点"""
algorithm = MyAlgorithm()
result = algorithm.execute(request.data)
if not result["success"]:
raise HTTPException(status_code=500, detail=result["error"])
return {
"request_id": request_id,
"status": "success",
"result": result["result"],
"metadata": result["metadata"]
}
步骤 4：定义请求模型
在 src/functional_scaffold/api/models.py 中添加：
class MyRequest(BaseModel):
"""我的请求模型"""
model_config = ConfigDict(
json_schema_extra={
"example": {"data": 10}
}
)
data: int = Field(..., description="输入数据")
步骤 5：测试
curl -X POST http://localhost:8000/my-endpoint \
-H "Content-Type: application/json" \
-d '{"data": 10}'
使用异步任务
对于耗时较长的算法，使用异步任务接口：
创建异步任务
curl -X POST http://localhost:8000/jobs \
-H "Content-Type: application/json" \
-d '{
"algorithm": "MyAlgorithm",
"params": {"data": 10},
"webhook": "https://your-callback-url.com/notify"
}'
响应：
{
"job_id": "a1b2c3d4e5f6",
"status": "pending",
"message": "任务已创建",
"created_at": "2026-02-02T10:00:00Z"
}
查询任务状态
curl http://localhost:8000/jobs/a1b2c3d4e5f6
响应：
{
"job_id": "a1b2c3d4e5f6",
"status": "completed",
"algorithm": "MyAlgorithm",
"result": {"input": 10, "output": 20},
"metadata": {"elapsed_time": 0.001}
}
本地开发技巧
查看 API 文档
启动服务后访问：
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc
运行测试
# 运行所有测试
pytest tests/ -v
# 运行单个测试文件
pytest tests/test_algorithms.py -v
代码格式化
# 格式化代码
black src/ tests/
# 检查代码规范
ruff check src/ tests/
下一步
常见问题
Q: 如何处理算法中的异常？
A: 在 process() 方法中抛出异常即可，框架会自动捕获并返回标准错误响应。
Q: 如何添加自定义指标？
A: 在 config/metrics.yaml 中定义指标，然后在代码中使用 incr() 或 observe() 记录。
Q: 如何访问外部服务（数据库、OSS）？
A: 在 config.py 中添加配置项，通过环境变量注入连接信息。
Q: 算法需要加载大模型文件怎么办？
A: 在算法类的 __init__ 方法中加载模型，框架会在容器启动时初始化。