# fastapi-template

**Repository Path**: tujiaw/fastapi-template

## Basic Information

- **Project Name**: fastapi-template
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-01-11
- **Last Updated**: 2026-01-16

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# FastAPI 最佳实践模板

基于 FastAPI 的企业级 Web 应用框架模板，遵循最佳实践。

## 特性

- 异步支持 (async/await)
- SQLAlchemy 2.0 + 异步数据库
- Pydantic v2 数据验证
- 自动 API 文档 (Swagger/ReDoc)
- CORS 支持
- 统一异常处理
- 请求日志中间件
- 配置管理
- 依赖注入
- 分页支持
- **JWT 认证（可配置开关）**
- **LangChain AI Agent 文章写作助手**
- 准备生产环境

## 项目结构

```
fastapi-template/
├── app/
│   ├── api/              # API 路由
│   │   └── v1/
│   │       ├── users.py
│   │       ├── health.py
│   │       └── __init__.py
│   ├── core/             # 核心模块
│   │   ├── config.py     # 配置
│   │   ├── deps.py       # 依赖注入
│   │   └── exceptions.py # 异常定义
│   ├── db/               # 数据库
│   │   └── database.py
│   ├── middleware/       # 中间件
│   │   └── logging_middleware.py
│   ├── models/           # 数据库模型
│   │   └── user.py
│   ├── schemas/          # Pydantic 模型
│   │   ├── user.py
│   │   ├── common.py
│   │   └── __init__.py
│   ├── services/         # 业务逻辑
│   │   └── user_service.py
│   ├── utils/            # 工具
│   │   └── logger.py
│   └── main.py           # 应用入口
├── tests/                # 测试
├── alembic/              # 数据库迁移
├── .env.example          # 环境变量示例
├── .gitignore
├── requirements.txt
└── README.md
```

## 快速开始

### 1. 安装依赖

```bash
# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate   # Windows

# 安装依赖
pip install -r requirements.txt
```

### 2. 配置环境变量

```bash
cp .env.example .env
# 编辑 .env 文件，修改配置
```

### 3. 运行应用

```bash
# 开发模式
uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

# 或者使用 Python
python -m app.main
```

### 4. 访问文档

- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc

## API 示例

### 健康检查

```bash
curl http://localhost:8000/api/v1/health
```

### 创建用户

```bash
curl -X POST http://localhost:8000/api/v1/users \
  -H "Content-Type: application/json" \
  -d '{
    "username": "testuser",
    "email": "test@example.com",
    "password": "password123"
  }'
```

### 获取用户列表

```bash
curl http://localhost:8000/api/v1/users?page=1&page_size=10
```

### 获取用户详情

```bash
curl http://localhost:8000/api/v1/users/1
```

## 认证功能

### 启用认证

在 `.env` 文件中设置：

```bash
ENABLE_AUTH=true
```

### 登录获取 Token

```bash
curl -X POST http://localhost:8000/api/v1/auth/login \
  -H "Content-Type: application/json" \
  -d '{
    "username": "admin",
    "password": "password123"
  }'
```

响应：

```json
{
  "access_token": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...",
  "token_type": "bearer",
  "expires_in": 1800,
  "user": {
    "id": 1,
    "username": "admin",
    "email": "admin@example.com"
  }
}
```

### 使用 Token 访问 API

```bash
curl http://localhost:8000/api/v1/users/me/protected \
  -H "Authorization: Bearer YOUR_TOKEN_HERE"
```

### 在 Swagger UI 中使用认证

1. 访问 http://localhost:8000/docs
2. 点击右上角的 "Authorize" 按钮
3. 输入 `your_token`（将 `your_token` 替换为实际的 token）
4. 点击 "Authorize" 确认

### 认证状态检查

```bash
curl http://localhost:8000/api/v1/auth/status
```

### 认证依赖使用

```python
from fastapi import Depends
from app.core.deps import get_current_user, get_current_user_optional

# 可选认证 - 如果启用认证且提供了 token 则验证，否则返回 None
async def public_endpoint(user: dict = Depends(get_current_user_optional)):
    if user:
        return {"message": f"Hello {user['username']}"}
    return {"message": "Hello anonymous"}

# 必须认证 - 如果启用认证则必须提供有效 token
async def protected_endpoint(user: dict = Depends(get_current_user)):
    return {"message": f"Hello {user['username']}"}
```

## 配置说明

主要配置项在 `.env` 文件中：

| 配置项 | 说明 | 默认值 |
|--------|------|--------|
| `DATABASE_URL` | 数据库连接字符串 | sqlite+aiosqlite:///./app.db |
| `SECRET_KEY` | JWT 密钥 | - |
| `ALGORITHM` | JWT 算法 | HS256 |
| `ACCESS_TOKEN_EXPIRE_MINUTES` | Token 过期时间（分钟） | 30 |
| `ENABLE_AUTH` | 是否启用认证 | 空（关闭） |
| `DEBUG` | 调试模式 | true |
| `CORS_ORIGINS` | 允许的跨域来源 | [] |

## 开发工具

```bash
# 代码格式化
black .

# 代码检查
ruff check .

# 类型检查
mypy app/

# 运行测试
pytest

# 测试覆盖率
pytest --cov=app tests/
```

## 数据库迁移

```bash
# 初始化 Alembic
alembic init alembic

# 创建迁移
alembic revision --autogenerate -m "description"

# 执行迁移
alembic upgrade head

# 回滚迁移
alembic downgrade -1
```

## 生产部署

### Docker

```dockerfile
FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
```

### 使用 Gunicorn

```bash
pip install gunicorn
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker
```

## 许可证

MIT License

## AI 文章写作助手（Human-in-the-Loop）

本项目集成了基于 LangChain 1.1+ 的 AI 文章写作助手，使用 `create_agent` 和 `HumanInTheLoopMiddleware` 实现真正的人机协同。

### 功能特性

- **LangChain Agent**：使用 LangChain 的 `create_agent` 构建智能写作助手
- **Human-in-the-Loop**：通过 `HumanInTheLoopMiddleware` 实现人工审核机制
- **工具调用审核**：AI 的每个工具调用都需要用户审阅后才能执行
- **多轮对话**：支持完整的对话历史和上下文管理
- **版本管理**：自动保存文章的每个修订版本
- **状态追踪**：实时追踪文章写作状态（写作中、审阅中、已完成、已拒绝）

### 技术架构

```
用户请求
    ↓
LangChain Agent (create_agent)
    ↓
LLM 生成响应
    ↓
HumanInTheLoopMiddleware (检查工具调用)
    ↓
触发中断 → 等待用户审阅
    ↓
用户决策 (approve/edit/reject)
    ↓
执行工具或重新生成
```

### 配置 LLM API

在 `.env` 文件中配置 LLM API 密钥（选择其中一个）：

```bash
# DeepSeek API Key (推荐)
DEEPSEEK_API_KEY=sk-...

# OpenAI API Key
OPENAI_API_KEY=sk-...

# Anthropic API Key
ANTHROPIC_API_KEY=sk-ant-...
```

### HITL API 端点

| 方法 | 端点 | 说明 |
|------|------|------|
| POST | `/api/v1/agent/article/sessions` | 创建新的写作会话 |
| POST | `/api/v1/agent/article/sessions/{id}/start` | 开始生成（触发 HITL） |
| POST | `/api/v1/agent/article/sessions/{id}/review-hitl` | 审阅并继续 |
| GET | `/api/v1/agent/article/sessions/{id}` | 获取会话详情 |
| GET | `/api/v1/agent/article/sessions/{id}/revisions` | 获取所有修订版本 |
| GET | `/api/v1/agent/article/sessions` | 获取会话列表 |

### 前端使用

启动 React 前端：

```bash
cd frontend
npm install
npm run dev
```

访问 `http://localhost:5173` 即可使用 AI 文章写作助手。

### 使用流程

1. **创建会话**：输入文章标题和主题要求
2. **AI 生成**：Agent 使用 `write_article` 工具生成内容
3. **触发审核**：`HumanInTheLoopMiddleware` 捕获工具调用并中断
4. **用户审阅**：
   - **同意 (approve)**：执行工具调用，完成写入
   - **编辑 (edit)**：修改内容后执行
   - **拒绝 (reject)**：拒绝执行，向 AI 提供反馈
5. **继续对话**：AI 根据反馈继续修改或完成任务

### API 示例

创建写作会话：

```bash
curl -X POST http://localhost:8000/api/v1/agent/article/sessions \
  -H "Content-Type: application/json" \
  -d '{
    "title": "人工智能的未来",
    "topic": "请写一篇关于人工智能未来发展趋势的文章，包括技术突破、应用场景和潜在挑战"
  }'
```

开始生成（触发 HITL）：

```bash
curl -X POST http://localhost:8000/api/v1/agent/article/sessions/1/start
```

审阅并同意：

```bash
curl -X POST http://localhost:8000/api/v1/agent/article/sessions/1/review-hitl \
  -H "Content-Type: application/json" \
  -d '{"decision_type": "approve"}'
```

审阅并编辑：

```bash
curl -X POST http://localhost:8000/api/v1/agent/article/sessions/1/review-hitl \
  -H "Content-Type: application/json" \
  -d '{
    "decision_type": "edit",
    "edited_content": "修改后的文章内容..."
  }'
```

审阅并拒绝：

```bash
curl -X POST http://localhost:8000/api/v1/agent/article/sessions/1/review-hitl \
  -H "Content-Type: application/json" \
  -d '{
    "decision_type": "reject",
    "feedback": "内容不够详细，请增加更多技术细节"
  }'
```

### HITL 工作原理

```python
# 创建 Agent 并添加 HITL 中间件
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=llm,
    tools=[write_article, revise_article],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_article": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                    "description": "文章内容已生成，请审阅"
                },
                "revise_article": {
                    "allowed_decisions": ["approve", "edit", "reject"],
                    "description": "文章修改建议已生成，请审阅"
                },
            },
        ),
    ],
    checkpointer=InMemorySaver(),
)

# 执行 Agent，遇到工具调用时会中断
result = agent.invoke(
    {"messages": [{"role": "user", "content": "写一篇文章..."}]},
    config={"configurable": {"thread_id": "session_123"}},
)

# 如果有中断，result 会包含 __interrupt__ 字段
if "__interrupt__" in result:
    # 展示给用户审阅
    # 用户决策后继续执行
    result = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config={"configurable": {"thread_id": "session_123"}},
    )
```

### 参考资源

- [LangChain Human-in-the-Loop 文档](https://docs.langchain.com/oss/python/langchain/human-in-the-loop)
- [LangChain Middleware 概述](https://docs.langchain.com/oss/python/langchain/middleware/overview)
- [LangGraph Checkpointer](https://langchain-ai.github.io/langgraph/concepts/persistence/)