跳到正文
GitHub·· 8 天前AI 评分22

基于 FastAPI + LangGraph 的 AI Agent 后端服务开源项目

XDC-666/fastapi-langgraph-agent-backend

AI 导读

开发者发布基于 FastAPI + LangGraph 的 AI Agent 后端服务,支持多轮对话、工具调用、RAG 知识库问答与 SSE 流式输出,可一键 Docker 部署。

正文

CI Python License: MIT

基于 FastAPI + LangGraph 的 AI Agent 后端服务。支持多轮对话、工具调用、RAG 知识库问答与流式输出(SSE),可一键 Docker 部署。

本项目为学习 / 面试作品,覆盖现代 AI 后端核心能力:FastAPI 异步服务、LangChain/LangGraph Agent 编排、PostgreSQL 持久化、Redis 缓存、RAG 检索增强生成。

🌐 简体中文 | English


✨ 功能特性

  • 🔐 JWT 用户认证:注册 / 登录 / 获取当前用户(bcrypt 加盐哈希)
  • 🛡️ 接口限流:基于 Redis 的固定窗口计数,登录按 IP 防暴力破解、对话按用户防刷成本
  • 🔒 登录失败锁定:同一账号连续失败达上限即锁定一段时间(Redis 不可用时自动降级放行)
  • 💬 多轮对话:基于 LangGraph 的 Agent 工作流,自动维护对话历史
  • 🛠️ 工具调用:内置计算器、网络搜索工具,模型可自主调用(ReAct 范式)
  • ⚡ 流式输出:SSE 逐 token 返回(LangGraph stream_mode="messages"),前端实时渲染
  • 📚 RAG 知识库:上传 PDF / Word / Markdown / TXT,自动切分 + 向量化 + 语义检索
  • 🗄️ PostgreSQL + Redis:关系型持久化 + 缓存 / 限流
  • 🧠 对话上下文窗口:只携带最近 N 条历史,避免长对话撑爆上下文 / 成本失控
  • 🧩 Alembic 数据库迁移:表结构可演进,改字段不再丢数据(替代裸 create_all)
  • 🧠 LangGraph Checkpointer:Agent 工作记忆持久化,多轮会话可恢复(可选 SqliteSaver)
  • 📊 Token 用量统计:记录每次对话的 prompt/completion token 并估算成本(面试加分项)
  • 🐳 Docker Compose 一键部署

🧱 技术栈

分类 技术
Web 框架 FastAPI(异步、自动 Swagger 文档)
Agent 编排 LangChain + LangGraph
数据库 PostgreSQL 16(SQLAlchemy 2.0 ORM)
缓存 Redis 7
认证 python-jose (JWT) + bcrypt(直接使用官方 API)
向量库 Chroma(本地持久化)
嵌入/对话 OpenAI / DeepSeek / 通义千问 兼容接口
部署 Docker + Docker Compose
测试 pytest

📐 架构图

flowchart LR
    Client[前端 / 客户端] -->|HTTP / SSE| API[FastAPI 路由层]
    API --> Auth[JWT 鉴权]
    API --> Chat[对话服务]
    API --> KB[知识库服务]
    Chat --> Agent[LangGraph Agent]
    Agent --> LLM[大模型 API]
    Agent --> Tools[工具: 计算器/搜索]
    Chat --> PG[(PostgreSQL 会话/消息)]
    KB --> Vec[(Chroma 向量库)]
    API --> Redis[(Redis 缓存)]
    PG --- Docker[(Docker Compose)]
    Redis --- Docker

Loading

Agent 工作流(LangGraph):

flowchart TD
    A[用户输入] --> B[Agent 节点: 大模型决策]
    B -->|需要工具| C[Tools 节点: 执行工具]
    C --> B
    B -->|直接回答| D[返回回复]

Loading


🚀 快速开始

方式一:Docker Compose(推荐)

# 1. 准备环境变量
cp .env.example .env
# 编辑 .env,至少填入 OPENAI_API_KEY(或兼容接口的 BASE_URL + KEY)

# 2. 一键启动(PostgreSQL + Redis + 后端)
docker compose up --build

# 3. 访问
# API 文档:   http://localhost:8000/docs
# 前端界面:   http://localhost:8000/app/(由 FastAPI 同源提供)

方式二:本地运行

# 1. 安装依赖(建议使用虚拟环境)
pip install -r requirements.txt

# 2. 启动本地 PostgreSQL 与 Redis(或用 Docker)
# 3. 配置 .env(DATABASE_URL / REDIS_URL / OPENAI_API_KEY)
cp .env.example .env

# 4. 启动服务
uvicorn app.main:app --reload --port 8000

🗄️ 数据库迁移(Alembic)

项目用 Alembic 管理表结构演进,不再依赖 create_all(后者改表结构会丢数据)。

# 安装迁移工具(已在 requirements.txt 中)
pip install -r requirements.txt

# 开发环境:可用 AUTO_CREATE_TABLES=true 让启动时自动建表,最省事
# 生产环境:AUTO_CREATE_TABLES=false,然后用迁移:

# 首次启动 / 升级到最新表结构
alembic upgrade head

# 改了 ORM 模型后,自动生成差异迁移脚本
alembic revision --autogenerate -m "描述这次变更"

# 查看迁移历史
alembic history

连接串来自 app.config.settings.DATABASE_URL,与项目其余部分共用一份配置。 Docker Compose 已在服务启动命令里自动执行 alembic upgrade head。


📡 API 一览

完整交互式文档见 http://localhost:8000/docs(Swagger UI)。

方法 路径 说明
POST /api/v1/auth/register 注册
POST /api/v1/auth/login 登录获取 JWT
GET /api/v1/auth/me 当前用户
POST /api/v1/conversations 新建会话
GET /api/v1/conversations 会话列表
GET /api/v1/conversations/{id}/messages 会话消息
POST /api/v1/chat 非流式对话
POST /api/v1/chat/stream 流式对话(SSE)
GET /api/v1/usage/me 当前用户 Token 用量与成本汇总
POST /api/v1/knowledge/upload 上传知识库文档
POST /api/v1/knowledge/ask 知识库检索
GET /api/v1/health 健康检查

调用示例:

# 注册
curl -X POST http://localhost:8000/api/v1/auth/register \
  -H "Content-Type: application/json" \
  -d '{"username":"alice","email":"alice@example.com","password":"secret123"}'

# 登录
TOKEN=$(curl -X POST http://localhost:8000/api/v1/auth/login \
  -d "username=alice&password=secret123" | python -c "import sys,json;print(json.load(sys.stdin)['access_token'])")

# 对话
curl -X POST http://localhost:8000/api/v1/chat \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"message":"1+2 等于几?"}'

📁 目录结构

fastapi-langgraph-agent-backend/
├── app/
│   ├── main.py              # FastAPI 入口
│   ├── config.py            # 配置
│   ├── database.py          # 数据库引擎
│   ├── dependencies.py      # 依赖注入(当前用户)
│   ├── models/              # ORM 模型(User/Conversation/Message/TokenUsage)
│   ├── schemas/             # Pydantic 模型
│   ├── routers/             # API 路由
│   ├── services/            # 业务逻辑
│   │   ├── agent/           # LangGraph Agent(graph / checkpointer)
│   │   ├── chat_service.py
│   │   ├── usage_service.py # Token 用量统计
│   │   └── knowledge_service.py
│   └── utils/               # 安全 / 缓存工具
├── alembic/                 # Alembic 迁移脚本
├── frontend/index.html      # 简易聊天前端
├── docker/                  # Dockerfile
├── tests/                   # pytest 测试
├── docker-compose.yml
├── alembic.ini
├── requirements.txt
└── README.md

🧠 学习收获

  • 用 FastAPI 构建异步 REST API,理解依赖注入、Pydantic 校验、中间件
  • 用 LangGraph 编排 Agent:state / node / conditional edge 的图编程思想
  • 实现 RAG:文档解析 → 切分 → 向量化 → 相似度检索 → 注入 prompt
  • 用 PostgreSQL + Redis 做数据持久化与缓存
  • 用 Docker Compose 实现可复现的一键部署

⚠️ 安全说明

  • .env 含密钥,已被 .gitignore 忽略,请勿提交
  • 生产环境请修改 SECRET_KEY
  • calculator 工具已做字符白名单校验,禁用任意代码执行

📄 License

MIT

来源:GitHub · github.com