Skip to content

shdancer/workflow-example

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

1 Commit
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Workflow Agent Template

基于 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

快速开始

1. 安装依赖

# 安装 uv(如果尚未安装)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 安装项目依赖
uv sync

2. 配置环境变量

cp .env.example .env
# 编辑 .env 文件,填入你的 API keys

需要配置的关键变量:

  • OPENAI_API_KEY: OpenAI API 密钥
  • WEATHER_API_KEY: OpenWeatherMap API 密钥(从 https://openweathermap.org/api 获取)
  • SMTP 配置(用于发送邮件)

3. 启动服务

uv run uvicorn src.api.app:app --reload

服务启动后访问:

使用示例

天气穿衣建议 API

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": []
}

添加新的 Workflow

步骤 1: 创建 Workflow 目录

mkdir -p src/workflows/my_workflow
touch src/workflows/my_workflow/__init__.py
touch src/workflows/my_workflow/workflow.py

步骤 2: 实现 Workflow

# 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()

步骤 3: 添加 API 路由

# 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)

步骤 4: 注册路由

# 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

License

MIT

About

No description, website, or topics provided.

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Languages