基于 FastAPI + LangGraph 的工作流/智能体项目模板,使用 uv 管理项目依赖。
.
├── src/
│ ├── api/ # FastAPI 应用层
│ │ ├── app.py # FastAPI 应用工厂
│ │ └── routes/ # API 路由
│ │ ├── __init__.py # 路由注册
│ │ └── weather.py # 天气 workflow 路由
│ ├── core/ # 核心模块
│ │ ├── config.py # 配置管理
│ │ └── workflow.py # Workflow 基类
│ ├── services/ # 外部服务集成
│ │ ├── weather.py # 天气 API 服务
│ │ └── email.py # 邮件服务
│ └── workflows/ # Workflow 实现
│ └── weather/ # 天气穿衣建议 workflow
│ └── workflow.py
├── .env.example # 环境变量示例
├── pyproject.toml # 项目配置
└── README.md
# 安装 uv(如果尚未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh
# 安装项目依赖
uv synccp .env.example .env
# 编辑 .env 文件,填入你的 API keys需要配置的关键变量:
OPENAI_API_KEY: OpenAI API 密钥WEATHER_API_KEY: OpenWeatherMap API 密钥(从 https://openweathermap.org/api 获取)- SMTP 配置(用于发送邮件)
uv run uvicorn src.api.app:app --reload服务启动后访问:
- API 文档: http://localhost:8000/docs
- 健康检查: http://localhost:8000/health
curl -X POST "http://localhost:8000/api/v1/weather/clothing-recommendation" \
-H "Content-Type: application/json" \
-d '{"city": "Beijing", "email_to": "your@email.com"}'响应示例:
{
"city": "Beijing",
"weather": {
"city": "Beijing",
"country": "CN",
"temperature": 22.5,
"feels_like": 21.8,
"humidity": 65,
"description": "多云",
"wind_speed": 3.2
},
"recommendation": "今天北京气温适中...",
"email_sent": true,
"success": true,
"errors": []
}mkdir -p src/workflows/my_workflow
touch src/workflows/my_workflow/__init__.py
touch src/workflows/my_workflow/workflow.py# src/workflows/my_workflow/workflow.py
from pydantic import BaseModel
from langgraph.graph import StateGraph, START, END
from langgraph.graph.state import CompiledStateGraph
from src.core.workflow import BaseWorkflow
class MyState(BaseModel):
"""Workflow 状态"""
input_data: str = ""
result: str = ""
class MyInput(BaseModel):
"""Workflow 输入"""
data: str
class MyOutput(BaseModel):
"""Workflow 输出"""
result: str
success: bool
class MyWorkflow(BaseWorkflow[MyState, MyInput, MyOutput]):
def build_graph(self) -> CompiledStateGraph:
graph = StateGraph(MyState)
graph.add_node("process", self._process)
graph.add_edge(START, "process")
graph.add_edge("process", END)
return graph.compile()
async def _process(self, state: MyState) -> dict:
# 实现处理逻辑
return {"result": f"Processed: {state.input_data}"}
async def run(self, input_data: MyInput) -> MyOutput:
initial_state = MyState(input_data=input_data.data)
final_state = await self.graph.ainvoke(initial_state)
return MyOutput(
result=final_state["result"],
success=True
)
my_workflow = MyWorkflow()# src/api/routes/my_workflow.py
from fastapi import APIRouter
from src.workflows.my_workflow.workflow import MyInput, MyOutput, my_workflow
router = APIRouter(prefix="/my-workflow", tags=["my-workflow"])
@router.post("/run", response_model=MyOutput)
async def run_workflow(input_data: MyInput) -> MyOutput:
return await my_workflow.run(input_data)# src/api/routes/__init__.py
from .my_workflow import router as my_workflow_router
api_router.include_router(my_workflow_router)- Python 3.12+: 使用最新语法特性
- uv: 快速的 Python 包管理器
- FastAPI: 高性能异步 Web 框架
- LangGraph: LLM 工作流编排框架
- LangChain: LLM 应用开发框架
- Pydantic: 数据验证和设置管理
- httpx: 异步 HTTP 客户端
- aiosmtplib: 异步 SMTP 客户端
# 安装开发依赖
uv sync --dev
# 运行测试
uv run pytest
# 代码格式化和检查
uv run ruff check src/
uv run ruff format src/| 变量名 | 说明 | 默认值 |
|---|---|---|
OPENAI_API_KEY |
OpenAI API 密钥 | - |
OPENAI_MODEL |
使用的模型 | gpt-4o-mini |
WEATHER_API_KEY |
OpenWeatherMap API 密钥 | - |
SMTP_HOST |
SMTP 服务器地址 | smtp.gmail.com |
SMTP_PORT |
SMTP 端口 | 587 |
SMTP_USERNAME |
SMTP 用户名 | - |
SMTP_PASSWORD |
SMTP 密码/应用密码 | - |
EMAIL_FROM |
发件人邮箱 | - |
EMAIL_TO |
默认收件人邮箱 | - |
APP_ENV |
运行环境 | development |
APP_DEBUG |
调试模式 | true |
MIT